Added first check for event-sourced application

Author: oskardudycz

samples/from_crud_to_eventsourcing/src/core/http.ts

@@ -8,11 +8,16 @@ import { assertUnsignedBigInt } from './validation';
 type WeakETag = `W/${string}`;
 type ETag = WeakETag | string;
 
+const enum HeaderNames {
+  IF_MATCH = 'if-match',
+  IF_NOT_MATCH = 'if-not-match',
+}
+
 export const WeakETagRegex = /W\/"(\d+.*)"/;
 
 export const enum ETagErrors {
   WRONG_WEAK_ETAG_FORMAT = 'WRONG_WEAK_ETAG_FORMAT',
-  MISSING_IF_MATCH_HEADER = 'MISSING_IF_MATCH_HEADER',
+  MISSING_HEADER = 'MISSING_HEADER',
 }
 
 export const isWeakETag = (etag: ETag): etag is WeakETag => {
@@ -29,18 +34,28 @@ export const toWeakETag = (value: unknown): WeakETag => {
   return `W/"${value}"`;
 };
 
-export const getETagFromIfMatch = (request: Request): ETag => {
-  const etag = request.headers['if-match'];
+export const getETagFromHeader = (
+  request: Request,
+  headerName: HeaderNames
+): ETag | undefined => {
+  const etag = request.headers[headerName];
 
-  if (etag === undefined) {
-    throw ETagErrors.MISSING_IF_MATCH_HEADER;
+  if (etag === undefined || etag.length > 0) {
+    return undefined;
   }
 
-  return etag;
+  return etag[0];
 };
 
-export const getWeakETagValueFromIfMatch = (request: Request): WeakETag => {
-  const etag = getETagFromIfMatch(request);
+export const getWeakETagValueFromHeader = (
+  request: Request,
+  headerName: HeaderNames
+): WeakETag | undefined => {
+  const etag = getETagFromHeader(request, headerName);
+
+  if (etag === undefined) {
+    return undefined;
+  }
 
   if (!isWeakETag(etag)) {
     throw ETagErrors.WRONG_WEAK_ETAG_FORMAT;
@@ -49,8 +64,32 @@ export const getWeakETagValueFromIfMatch = (request: Request): WeakETag => {
   return getWeakETagValue(etag);
 };
 
-export const getExpectedRevisionFromETag = (request: Request): bigint =>
-  assertUnsignedBigInt(getWeakETagValueFromIfMatch(request));
+export const getExpectedRevision = (
+  request: Request,
+  headerName: HeaderNames
+): bigint | undefined => {
+  const eTag = getWeakETagValueFromHeader(request, headerName);
+
+  if (eTag === undefined) {
+    return undefined;
+  }
+
+  return assertUnsignedBigInt(eTag);
+};
+
+export const getExpectedRevisionFromIfMatch = (request: Request): bigint => {
+  const revision = getExpectedRevision(request, HeaderNames.IF_MATCH);
+
+  if (revision === undefined) {
+    throw ETagErrors.MISSING_HEADER;
+  }
+
+  return revision;
+};
+
+export const getExpectedRevisionFromIfNotMatch = (
+  request: Request
+): bigint | undefined => getExpectedRevision(request, HeaderNames.IF_NOT_MATCH);
 
 //////////////////////////////////////
 /// HTTP Helpers

samples/from_crud_to_eventsourcing/src/core/postgres.ts

@@ -41,12 +41,16 @@ export const disconnectFromPostgres = async () => {
 export const runPostgresMigration = ({
   connectionString,
   migrationsPath,
+  ignoreEdditedMigrationFile = false,
 }: {
   connectionString: string;
   migrationsPath: string;
+  ignoreEdditedMigrationFile?: boolean;
 }) => {
   execSync(
-    `npx node_modules/@databases/pg-migrations apply --database ${connectionString} --directory ${migrationsPath}`
+    `npx node_modules/@databases/pg-migrations apply --database ${connectionString} --directory ${migrationsPath} ${
+      ignoreEdditedMigrationFile ? '--ignore-error migration_file_edited' : ''
+    }`
   );
 };
 

samples/from_crud_to_eventsourcing/src/core/validation.ts

@@ -39,6 +39,10 @@ export const assertPositiveNumberOrUndefined = (
 };
 
 export const assertUnsignedBigInt = (value: string): bigint => {
+  if (value === undefined) {
+    throw ValidationErrors.NOT_AN_UNSIGNED_BIGINT;
+  }
+
   const number = BigInt(value);
   if (number < 0) {
     throw ValidationErrors.NOT_AN_UNSIGNED_BIGINT;

samples/from_crud_to_eventsourcing/src/crud/index.ts

@@ -1,14 +1,12 @@
 import { startAPI } from '#core/api';
 import { disconnectFromPostgres } from '#core/postgres';
-import { disconnectFromEventStore } from '#eventsourced/core/streams';
 import { router } from './shoppingCarts/routes';
 
 //////////////////////////////////////////////////////////
 /// Make sure that we dispose Postgres connection pool
 //////////////////////////////////////////////////////////
 
 process.once('SIGTERM', disconnectFromPostgres);
-process.once('SIGTERM', disconnectFromEventStore);
 
 //////////////////////////////////////////////////////////
 /// API

samples/from_crud_to_eventsourcing/src/eventsourced/core/subscriptions.ts

@@ -20,7 +20,7 @@ export type EventHandler = (event: SubscriptionResolvedEvent) => Promise<void>;
 
 export const SubscriptionToAll =
   (
-    eventStore: EventStoreDBClient,
+    getEventStore: () => EventStoreDBClient,
     loadCheckpoint: (subscriptionId: string) => Promise<bigint | undefined>
   ) =>
   async (subscriptionId: string, handlers: EventHandler[]) => {
@@ -29,7 +29,7 @@ export const SubscriptionToAll =
       ? START
       : { prepare: currentPosition, commit: currentPosition };
 
-    const subscription = eventStore.subscribeToAll({
+    const subscription = getEventStore().subscribeToAll({
       fromPosition,
       filter: excludeSystemEvents(),
     });
@@ -98,6 +98,6 @@ export const handleEventInPostgresTransactionScope =
   };
 
 export const SubscriptionToAllWithPostgresCheckpoints = SubscriptionToAll(
-  getEventStore(),
+  getEventStore,
   loadCheckPointFromPostgres
 );