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

import { PrintfulClient, ProductSyncer, WebflowClient, type ProductSyncerOptions } from "@blade-and-brawn/commerce";
import { env, log } from "../util";
import { db } from "../database/db";
import { DatabaseError } from "pg";
import type { Insertable, Selectable } from "kysely";
import type { ProductSyncs } from "../database/out/db";

type SyncStatus = "queued" | "active" | "failed" | "succeeded";

type QueuedSync = {
    id?: string,
    session: {
        name: string,
        id: string
    },
    filter: ProductSyncerOptions["filter"]
}

class AlreadyClaimedError extends Error { }

export class CommerceService {
    public readonly printful: PrintfulClient;
    public readonly webflow: WebflowClient;
    private readonly productSyncer: ProductSyncer;
    readonly queue: SyncQueue;

    constructor() {
        this.printful = new PrintfulClient({
            token: env.PRINTFUL_AUTH,
            storeId: env.PRINTFUL_STORE_ID,
        });
        this.webflow = new WebflowClient({
            siteId: env.WEBFLOW_SITE_ID,
            collectionsId: env.WEBFLOW_COLLECTION_ID,
            token: env.WEBFLOW_AUTH,
            webhookSecret: env.WEBFLOW_WEBHOOK_SECRET,
        });
        this.productSyncer = new ProductSyncer(this.printful, this.webflow, log);
        this.queue = new SyncQueue();
    }

    async sync(queuedSync: QueuedSync) {
        let id = queuedSync.id ?? null;
        let freedSlot = false;
        try {
            log.info({ session: queuedSync.session }, "starting sync run");
            await this.productSyncer.sync({
                filter: queuedSync.filter,
                onStart: async (startDate) => {
                    const values: Insertable<ProductSyncs> = {
                        session_id: queuedSync.session.id,
                        session_name: queuedSync.session.name,
                        started_at: startDate,
                        printful_product_id_filter: queuedSync.filter?.printfulProductIds ?? null,
                    };
                    if (id) {
                        const result = await db.updateTable("product_syncs")
                            .set(values)
                            .where("id", "=", id)
                            .where("started_at", "is", null)
                            .executeTakeFirst();

                        // Guard against another worker/machine having already claimed
                        // this queued row between `queue.front()` and now.
                        if (result.numUpdatedRows === 0n) throw new AlreadyClaimedError();
                    }
                    else {
                        const result = await db.insertInto("product_syncs")
                            .values(values)
                            .returning("id")
                            .executeTakeFirstOrThrow();
                        id = result.id;
                    }
                },
                onStep: async (printfulProductIds: number[]) => {
                    if (!id) throw new Error("Expected sync run ID");
                    await db.updateTable("product_syncs")
                        .set({ syncing_printful_product_ids: printfulProductIds })
                        .where("id", "=", id)
                        .execute();
                },
                onCompletion: async (completionDate, duration) => {
                    if (!id) throw new Error("Expected sync run ID");
                    await db.updateTable("product_syncs")
                        .set({ syncing_printful_product_ids: [], ended_at: completionDate })
                        .where("id", "=", id)
                        .execute();
                    freedSlot = true;
                }
            });
        }
        catch (err) {
            if (err instanceof AlreadyClaimedError) {
                log.info("sync already claimed by another worker, skipping");
                return;
            }

            const activeSync = await this.getActiveSync();

            const activeSyncConflict = activeSync && err instanceof DatabaseError && err.code === "23505";
            if (activeSyncConflict) {
                // If active sync has been syncing for 10 min (600000ms), clear it and retry this sync
                if (Date.now() - (activeSync.started_at?.getTime() ?? 0) > 600000) {
                    log.info("already active sync exceeded timeout");
                    await db.updateTable("product_syncs")
                        .set({ syncing_printful_product_ids: [], ended_at: new Date(), has_failed: true })
                        .where("id", "=", activeSync.id)
                        .execute();
                    await this.sync(queuedSync);
                }
                else {
                    log.info("sync already active, enqueuing next sync");
                    await this.queue.enqueue(queuedSync);
                }
            }
            else {
                if (id) {
                    await db.updateTable("product_syncs")
                        .set({ syncing_printful_product_ids: [], ended_at: new Date(), has_failed: true })
                        .where("id", "=", id)
                        .execute();
                    freedSlot = true;
                }
                throw err;
            }
        }
        finally {
            if (freedSlot) {
                try {
                    const frontQueuedSync = await this.queue.front();
                    if (frontQueuedSync) await this.sync(frontQueuedSync);
                }
                catch (err) {
                    log.error({ err }, "failed to process next queued sync")
                }
            }
        }
    }

    async getActiveSync(sessionId?: string) {
        return await db.selectFrom("product_syncs")
            .selectAll()
            .$if(sessionId !== undefined, (qb) =>
                qb.where("session_id", "=", sessionId!)
            )
            .where("started_at", "is not", null)
            .where("ended_at", "is", null)
            .limit(1)
            .executeTakeFirst();
    }

    async getLatestSync(sessionId?: string) {
        return await db.selectFrom("product_syncs")
            .selectAll()
            .$if(sessionId !== undefined, (qb) =>
                qb.where("session_id", "=", sessionId!)
            )
            .orderBy("created_at", "desc")
            .limit(1)
            .executeTakeFirst();
    }

    deriveSyncStatus(sync: Selectable<ProductSyncs>): SyncStatus {
        if (!sync.started_at) return "queued";
        if (!sync.ended_at) return "active";
        return sync.has_failed ? "failed" : "succeeded";
    }
}

class SyncQueue {
    async enqueue(queuedSync: QueuedSync) {
        const values: Insertable<ProductSyncs> = {
            session_id: queuedSync.session.id,
            session_name: queuedSync.session.name,
            printful_product_id_filter: queuedSync.filter?.printfulProductIds ?? null
        };
        if (queuedSync.id) {
            await db.updateTable("product_syncs")
                .set(values)
                .where("id", "=", queuedSync.id)
                .execute();
        }
        else {
            await db.insertInto("product_syncs")
                .values(values)
                .execute();
        }
    }

    async front(): Promise<QueuedSync | undefined> {
        const result = await db.selectFrom("product_syncs")
            .selectAll()
            .where("started_at", "is", null)
            .orderBy("created_at", "asc")
            .limit(1)
            .executeTakeFirst();
        if (!result) return;

        return {
            id: result.id,
            session: {
                id: result.session_id,
                name: result.session_name
            },
            filter: {
                printfulProductIds: result.printful_product_id_filter
            }
        }
    }
}