platform/apps/api/src/services/events.ts

import type { Static } from "@sinclair/typebox";
import { t } from "elysia";
import { sql, type Selectable, type Transaction } from "kysely";
import type { DB, JsonValue } from "../database/out/db";
import { db } from "../database/db";
import { log } from "../util";
import { minToMs } from "@blade-and-brawn/domain";

export const EventStatusSchema = t.Union([
    t.Literal("waiting"),
    t.Literal("available"),
    t.Literal("processing"),
    t.Literal("failed"),
    t.Literal("resolved"),
]);
export type EventStatus = Static<typeof EventStatusSchema>;

export type EventSource = "portal" | "printful" | "webflow";

export class EventsService {
    static readonly CONCURRENCY_TIMEOUT_MS = minToMs(10);

    private _lastCleanDate: Date = new Date(0);
    get lastCleanDate() { return this._lastCleanDate; };

    private static 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);
    }

    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" : "resolved";
    }

    async get(id: string, trx?: Transaction<DB>) {
        return await (trx ?? db).selectFrom("events")
            .selectAll()
            .where("id", "=", id)
            .executeTakeFirst();
    }

    async list(opt: {
        filter?: { status?: EventStatus, type?: string, group?: string },
        limit?: number,
        offset?: number,
    } = {}, trx?: Transaction<DB>) {
        return await (trx ?? db).selectFrom("events")
            .select([
                "id", "group", "type", "source", "started_at", "ended_at", "available_at",
                "errors", "rate_limits", "has_failed", "last_error", "created_at",
            ])
            .$if(opt.filter?.group !== undefined, (qb) => qb.where("group", "=", opt.filter!.group!))
            .$if(opt.filter?.type !== undefined, (qb) => qb.where("type", "=", opt.filter!.type!))
            .$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 === "resolved", (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")
            .orderBy("id", "desc")
            .$if(opt.limit !== undefined, (qb) => qb.limit(opt.limit!))
            .$if(opt.offset !== undefined, (qb) => qb.offset(opt.offset!))
            .execute();
    }

    async setState(id: string, state: JsonValue) {
        await db.updateTable("events")
            .set({ state })
            .where("id", "=", id)
            .execute();
    }

    async setLastError(id: string, err: unknown) {
        await db.updateTable("events")
            .set({ last_error: EventsService.serializeError(err) })
            .where("id", "=", id)
            .execute();
    }

    async fail(id: string, trx?: Transaction<DB>): Promise<void> {
        await (trx ?? db).updateTable("events")
            .set((eb) => ({
                state: null,
                ended_at: new Date(),
                has_failed: true,
                errors: eb("errors", "+", 1)
            }))
            .where("id", "=", id)
            .execute();
    }

    async resolve(id: string, trx?: Transaction<DB>): Promise<void> {
        await (trx ?? db).updateTable("events")
            .set({ state: null, ended_at: new Date(), last_error: null })
            .where("id", "=", id)
            .execute();
    }

    async retry(id: string): Promise<boolean> {
        const result = await db.updateTable("events")
            .set({
                started_at: null,
                ended_at: null,
                has_failed: false,
                available_at: new Date(),
            })
            .where("id", "=", id)
            .where("has_failed", "=", true)
            .executeTakeFirst();
        return result.numUpdatedRows > 0n;
    }

    async clean(opt: { filter?: { type?: string, group?: string } } = {}) {
        try {
            const timedOutEvents = await db.updateTable("events")
                .set((eb) => ({
                    state: null,
                    ended_at: new Date(),
                    has_failed: true,
                    errors: eb("errors", "+", 1),
                }))
                .where("started_at", "is not", null)
                .where("ended_at", "is", null)
                .where("started_at", "<", new Date(Date.now() - EventsService.CONCURRENCY_TIMEOUT_MS))
                .$if(opt.filter?.type !== undefined, (qb) => qb.where("type", "=", opt.filter!.type!))
                .$if(opt.filter?.group !== undefined, (qb) => qb.where("group", "=", opt.filter!.group!))
                .returning(["id"])
                .execute();

            for (const event of timedOutEvents) {
                log.warn({ name: event.id }, "processing event timed out, failed it");
            }
        }
        finally {
            this._lastCleanDate = new Date();
        }
    }
}