feat(balance-worker): add partition track writer

Author: og2701

apps/balance-worker/src/state/sqliteBalanceStateStore.ts

@@ -41,6 +41,11 @@ export type KafkaRecordPosition = {
 	offset: bigint;
 };
 
+export type DurableTrackOutcomeRecord = {
+	position: KafkaRecordPosition;
+	outcome: TrackOutcome;
+};
+
 export type DurableTrackOutcomeApplyResult =
 	| {
 			kind: "applied" | "duplicate";
@@ -174,98 +179,121 @@ export class SqliteBalanceStateStore {
 	applyDurableTrackOutcome({
 		position,
 		outcome,
+	}: DurableTrackOutcomeRecord): DurableTrackOutcomeApplyResult {
+		const [result] = this.applyDurableTrackOutcomes({
+			records: [{ position, outcome }],
+		});
+		if (!result) throw new Error("Expected a durable track outcome result");
+		return result;
+	}
+
+	applyDurableTrackOutcomes({
+		records,
 	}: {
-		position: KafkaRecordPosition;
-		outcome: TrackOutcome;
-	}): DurableTrackOutcomeApplyResult {
-		assertTopic({ topic: position.topic });
-		assertPartition({ partition: position.partition });
-		assertOffset({ offset: position.offset });
-		const parsedOutcome = parseTrackOutcome({ input: outcome });
+		records: readonly DurableTrackOutcomeRecord[];
+	}): DurableTrackOutcomeApplyResult[] {
+		const parsedRecords = records.map(({ position, outcome }) => {
+			assertTopic({ topic: position.topic });
+			assertPartition({ partition: position.partition });
+			assertOffset({ offset: position.offset });
+			return {
+				position,
+				outcome: parseTrackOutcome({ input: outcome }),
+			};
+		});
+		if (parsedRecords.length === 0) return [];
 
 		return this.database
-			.transaction(() => {
-				const expectedOffset = readNextOffset({
-					database: this.database,
-					topic: position.topic,
-					partition: position.partition,
-				});
-				if (expectedOffset === null) {
-					throw new PartitionProgressNotFoundError({
-						topic: position.topic,
-						partition: position.partition,
-					});
-				}
-				if (position.offset < expectedOffset) {
-					return {
-						kind: "position_already_applied",
-						nextOffset: expectedOffset,
-					} as const;
-				}
-				const partitionKey = meteringPartitionKeyOf({
-					identity: parsedOutcome.identity,
-				});
-				const state = readState({
-					database: this.database,
-					identity: parsedOutcome.identity,
-				});
-				if (!state) throw new MeteringStateNotFoundError({ partitionKey });
+			.transaction(() =>
+				parsedRecords.map(({ position, outcome }) =>
+					this.applyParsedDurableTrackOutcome({ position, outcome }),
+				),
+			)
+			.immediate();
+	}
 
-				const existingReceipt = readTrackReceipt({
-					database: this.database,
-					identity: parsedOutcome.identity,
-					commandId: parsedOutcome.commandId,
-				});
-				const executed = executeEngineTrack({
-					state,
-					outcome: parsedOutcome,
-					existingReceipt,
-				});
-				const nextOffset = position.offset + 1n;
+	private applyParsedDurableTrackOutcome({
+		position,
+		outcome,
+	}: DurableTrackOutcomeRecord): DurableTrackOutcomeApplyResult {
+		const expectedOffset = readNextOffset({
+			database: this.database,
+			topic: position.topic,
+			partition: position.partition,
+		});
+		if (expectedOffset === null) {
+			throw new PartitionProgressNotFoundError({
+				topic: position.topic,
+				partition: position.partition,
+			});
+		}
+		if (position.offset < expectedOffset) {
+			return {
+				kind: "position_already_applied",
+				nextOffset: expectedOffset,
+			};
+		}
 
-				if (executed.kind === "applied") {
-					const stateUpdate = updateState({
-						database: this.database,
-						partitionKey,
-						revisionBefore: parsedOutcome.revisionBefore,
-						state: executed.state,
-					});
-					if (stateUpdate.changes !== 1) {
-						throw new CorruptBalanceStateError({ partitionKey });
-					}
+		const partitionKey = meteringPartitionKeyOf({ identity: outcome.identity });
+		const state = readState({
+			database: this.database,
+			identity: outcome.identity,
+		});
+		if (!state) throw new MeteringStateNotFoundError({ partitionKey });
 
-					insertTrackReceipt({
-						database: this.database,
-						partitionKey,
-						position,
-						receipt: executed.receipt,
-					});
-				}
+		const existingReceipt = readTrackReceipt({
+			database: this.database,
+			identity: outcome.identity,
+			commandId: outcome.commandId,
+		});
+		const executed = executeEngineTrack({
+			state,
+			outcome,
+			existingReceipt,
+		});
+		const nextOffset = position.offset + 1n;
 
-				const progressUpdate = advancePartitionProgress({
-					database: this.database,
-					topic: position.topic,
-					partition: position.partition,
-					expectedOffset,
-					nextOffset,
-				});
-				if (progressUpdate.changes !== 1) {
-					throw new UnexpectedKafkaOffsetError({
-						topic: position.topic,
-						partition: position.partition,
-						expectedOffset,
-						receivedOffset: position.offset,
-					});
-				}
+		if (executed.kind === "applied") {
+			const stateUpdate = updateState({
+				database: this.database,
+				partitionKey,
+				revisionBefore: outcome.revisionBefore,
+				state: executed.state,
+			});
+			if (stateUpdate.changes !== 1) {
+				throw new CorruptBalanceStateError({ partitionKey });
+			}
 
-				return {
-					kind: executed.kind,
-					state: executed.state,
-					receipt: executed.receipt,
-					nextOffset,
-				} as const;
-			})
-			.immediate();
+			insertTrackReceipt({
+				database: this.database,
+				partitionKey,
+				position,
+				receipt: executed.receipt,
+			});
+		}
+
+		const progressUpdate = advancePartitionProgress({
+			database: this.database,
+			topic: position.topic,
+			partition: position.partition,
+			expectedOffset,
+			nextOffset,
+		});
+		if (progressUpdate.changes !== 1) {
+			throw new UnexpectedKafkaOffsetError({
+				topic: position.topic,
+				partition: position.partition,
+				expectedOffset,
+				receivedOffset: position.offset,
+			});
+		}
+
+		return {
+			kind: executed.kind,
+			state: executed.state,
+			receipt: executed.receipt,
+			nextOffset,
+		};
 	}
 
 	close(): void {

apps/balance-worker/src/writer/committedTrackOutcomeAppender.ts

@@ -0,0 +1,20 @@
+import type { TrackOutcome } from "@autumn/balance-engine";
+
+export class TrackOutcomeBatchNotCommittedError extends Error {
+	constructor({ cause }: { cause: unknown }) {
+		super("Track outcome batch was not committed", { cause });
+		this.name = "TrackOutcomeBatchNotCommittedError";
+	}
+}
+
+export type CommittedTrackOutcomeAppender = {
+	appendCommitted({
+		topic,
+		partition,
+		outcomes,
+	}: {
+		topic: string;
+		partition: number;
+		outcomes: readonly TrackOutcome[];
+	}): Promise<{ baseOffset: bigint }>;
+};