Software/platform
1
1
Fork 0
Code Issues 3 Pull Requests Actions Packages Projects Releases Wiki Activity

Implement events api routes

Browse Source
This commit is contained in:
Dominic Ferrando
2026-07-16 00:55:04 -04:00
parent ee9b03796d
commit 48765b4b89
6 changed files with 225 additions and 97 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/api/src/services/event-queue.ts
+39 -85
View File
@@ -1,14 +1,12 @@
import { sql, type Selectable, type Transaction } from "kysely";
import { type Selectable, type Transaction } from "kysely";
import type { DB, JsonValue } from "../database/out/db";
import { log } from "../util";
import { sleep } from "bun";
import { db } from "../database/db";
import { AssertError, Value } from "@sinclair/typebox/value";
import type { TSchema, Static } from "@sinclair/typebox";
import { minToMs, RateLimitError, secToMs } from "@blade-and-brawn/domain";
export type EventSource = "portal" | "printful" | "webflow";
export type EventStatus = "waiting" | "available" | "processing" | "failed" | "fulfilled";
import { RateLimitError, secToMs } from "@blade-and-brawn/domain";
import { EventsService, type EventSource } from "./events";
type RetryType = "error" | "ratelimit";
@@ -27,7 +25,6 @@ type EventQueueCfg = {
},
concurrency: {
limit: number,
timeoutMs: number,
},
};
@@ -47,32 +44,33 @@ function exponentialDelay(initialDelay: number, tries: number, cap: number = Inf
return Math.min(initialDelay * (2 ** tries), cap);
}
function serializeError(err: unknown): JsonValue {
if (err instanceof Error) {
const payload = "payload" in err ? { payload: err.payload as JsonValue } : {};
return { name: err.name, message: err.message, ...payload };
}
return String(err);
}
export class EventQueue<G extends string, T extends Record<string, EventTypeOptions<any, any>>> {
private static readonly DEQUEUE_RETRY_DELAY_MS = secToMs(10);
private static readonly DEQUEUE_RETRY_LIMIT = 3;
private static readonly RATE_LIMIT_RETRY_LIMIT = 15;
private static readonly JITTER_FACTOR = 0.2;
private _lastCleanDate: Date = new Date(0);
readonly cfg: EventQueueCfg;
readonly group: G;
readonly events: T;
readonly types: T;
constructor(opt: { group: G, cfg: EventQueueCfg, events: T }) {
this.cfg = opt.cfg;
this.group = opt.group;
this.events = opt.events;
}
private readonly Events = new EventsService();
private _lastCleanDate: Date = new Date(0);
get lastCleanDate() { return this._lastCleanDate; };
static status(event: Pick<Selectable<DB["events"]>, "available_at" | "started_at" | "ended_at" | "has_failed">): EventStatus {
if (!event.started_at) {
if (Date.now() < event.available_at.getTime()) return "waiting";
return "available";
};
if (!event.ended_at) return "processing";
return event.has_failed ? "failed" : "fulfilled";
constructor(opt: { group: G, cfg: EventQueueCfg, types: T }) {
this.cfg = opt.cfg;
this.group = opt.group;
this.types = opt.types;
}
async drain(): Promise<void> {
@@ -160,37 +158,6 @@ export class EventQueue<G extends string, T extends Record<string, EventTypeOpti
return result.id;
}
async fail(id: string, trx?: Transaction<DB>): Promise<void> {
await (trx ?? db).updateTable("events")
.set({ state: null, ended_at: new Date(), has_failed: true })
.where("id", "=", id)
.execute();
}
async fulfill(id: string, trx?: Transaction<DB>): Promise<void> {
await (trx ?? db).updateTable("events")
.set({ state: null, ended_at: new Date() })
.where("id", "=", id)
.execute();
}
async clean() {
try {
// Find and fail any timed out processing events
const processingEvents = await this.list({ filter: { status: "processing" } });
for (const processingEvent of processingEvents) {
const processingEventTimedOut = (Date.now() - (processingEvent.started_at?.getTime() ?? 0) > this.cfg.concurrency.timeoutMs);
if (processingEventTimedOut) {
log.warn({ name: processingEvent.id }, "processing event has timed out. failing it and continuing...");
await this.fail(processingEvent.id);
}
}
}
finally {
this._lastCleanDate = new Date();
}
}
async front(trx?: Transaction<DB>) {
const result = await (trx ?? db).selectFrom("events")
.select(["id", "type", "payload", "available_at", "errors", "rate_limits"])
@@ -203,48 +170,28 @@ export class EventQueue<G extends string, T extends Record<string, EventTypeOpti
return result;
}
async list(opt: { filter?: { status?: EventStatus } } = {}, trx?: Transaction<DB>) {
return await (trx ?? db).selectFrom("events")
.select(["started_at", "id"])
.where("group", "=", this.group)
.$if(opt.filter?.status === "waiting", (qb) => qb
.where("started_at", "is", null)
.where("available_at", ">", new Date())
)
.$if(opt.filter?.status === "available", (qb) => qb
.where("started_at", "is", null)
.where("available_at", "<=", new Date())
)
.$if(opt.filter?.status === "processing", (qb) => qb
.where("started_at", "is not", null)
.where("ended_at", "is", null)
)
.$if(opt.filter?.status === "failed", (qb) => qb
.where("started_at", "is not", null)
.where("ended_at", "is not", null)
.where("has_failed", "is", true)
)
.$if(opt.filter?.status === "fulfilled", (qb) => qb
.where("started_at", "is not", null)
.where("ended_at", "is not", null)
.where("has_failed", "is", false)
)
.orderBy(sql`(started_at is not null and ended_at is null)`, "desc")
.orderBy("created_at", "desc")
.execute();
async clean(opt: { filter?: { type?: string } } = {}) {
try {
await this.Events.clean({ filter: { group: this.group, ...opt.filter } });
}
finally {
this._lastCleanDate = new Date();
}
}
private async processEvent(event: Pick<Selectable<DB["events"]>, "id" | "type" | "errors" | "payload" | "rate_limits">) {
log.info({ id: event.id, type: event.type, payload: event.payload }, "processing event");
if (!(event.type in this.events)) throw new Error(`event type not registered in this queue (${this.group}): ${event.type}`);
const cfg = this.events[event.type as keyof T]!;
if (!(event.type in this.types)) throw new Error(`event type not registered in this queue (${this.group}): ${event.type}`);
const cfg = this.types[event.type as keyof T]!;
try {
Value.Assert(cfg.schemas.payload, event.payload);
await cfg.processor(event.payload, (state) => this.setEventState(event.id, state), event.id);
await this.fulfill(event.id);
await this.Events.fulfill(event.id);
}
catch (err) {
await this.setLastError(event.id, err);
if (err instanceof RateLimitError && event.rate_limits < EventQueue.RATE_LIMIT_RETRY_LIMIT) {
log.warn(
{ rateLimits: event.rate_limits, delay: err.retryDelayMs, source: err.source, payload: err.payload },
@@ -256,7 +203,7 @@ export class EventQueue<G extends string, T extends Record<string, EventTypeOpti
if (err instanceof AssertError) {
log.error({ errors: event.errors }, "could not process event: malformed payload");
await this.fail(event.id);
await this.Events.fail(event.id);
return;
}
@@ -271,7 +218,7 @@ export class EventQueue<G extends string, T extends Record<string, EventTypeOpti
}
log.error({ err, errors: event.errors, rateLimits: event.rate_limits }, "event failed");
await this.fail(event.id);
await this.Events.fail(event.id);
}
}
@@ -282,6 +229,13 @@ export class EventQueue<G extends string, T extends Record<string, EventTypeOpti
.execute();
}
private async setLastError(id: string, err: unknown) {
await db.updateTable("events")
.set({ last_error: serializeError(err) })
.where("id", "=", id)
.execute();
}
private async retry(id: string, delayMs: number, type: RetryType) {
await db.updateTable("events")
.set((eb) => ({