fix(smartsheet): register sync cron and fix backfill transaction visibility
Deploy to route.crispygoat.com / deploy (push) Successful in 4m9s
Deploy to route.crispygoat.com / deploy (push) Successful in 4m9s
Two bugs were silently breaking every Smartsheet sync since launch:
1. The /api/water-log/smartsheet-sync route was never registered in
vercel.json's crons array. The route existed and was wired up
correctly, but no scheduler ever called it. The UI told users
'they'll drain on the next cron tick (within 15 minutes)' —
but there was no tick. Queue rows accumulated indefinitely.
Fix: register the route with schedule '*/15 * * * *'.
2. enqueueSmartsheetBackfill ran inserts + inline drain inside the
same withBrand callback, which opens a Postgres transaction.
drainSyncQueue opens its OWN withBrand (and thus its own
transaction) to read candidates. Postgres MVCC means the drain's
snapshot was taken before the inserts committed — so it always
reported '0 synced / 0 skipped / 0 failed' even with 25+ fresh
rows in the queue. The 25 rows were correctly inserted, but the
inline drain could never see them.
Fix: split the backfill into three explicit phases, each in its
own transaction:
- Phase 1: validate + query + insert queue rows (commits)
- Phase 2: inline drain in a FRESH transaction (sees phase 1)
- Phase 3: audit log (logAuditEvent opens its own withBrand)
The phase 1 callback returns a discriminated union so config-
missing / sync-disabled / empty-result paths still short-circuit
cleanly without ever touching phase 2.
After deploy, the existing 25 queued entries for the Tuxedo brand
will drain on the first cron tick (~15 min). To drain immediately
without waiting:
curl -H 'Authorization: Bearer $SMARTSHEET_CRON_SECRET' \
-X POST http://crispygoat.com:3100/api/water-log/smartsheet-sync
This commit is contained in:
@@ -523,8 +523,14 @@ export async function enqueueSmartsheetBackfill(
|
|||||||
return { success: false, error: auth.error, ...empty };
|
return { success: false, error: auth.error, ...empty };
|
||||||
}
|
}
|
||||||
|
|
||||||
return withBrand(brandId, async (db) => {
|
// ── Phase 1: validate + insert queue rows (own transaction) ──
|
||||||
// 1. Require active config.
|
// IMPORTANT: the drain (phase 2) MUST run in a separate transaction
|
||||||
|
// so it can see the inserts we just committed. The previous
|
||||||
|
// implementation did both inside one withBrand callback, which
|
||||||
|
// meant the drain's MVCC snapshot was taken before the inserts
|
||||||
|
// committed — so it always reported 0/0/0 even with 25+ fresh
|
||||||
|
// rows in the queue.
|
||||||
|
const phase1 = await withBrand(brandId, async (db) => {
|
||||||
const configRows = await db
|
const configRows = await db
|
||||||
.select({
|
.select({
|
||||||
syncEnabled: waterSmartsheetConfig.syncEnabled,
|
syncEnabled: waterSmartsheetConfig.syncEnabled,
|
||||||
@@ -535,26 +541,19 @@ export async function enqueueSmartsheetBackfill(
|
|||||||
const config = configRows[0];
|
const config = configRows[0];
|
||||||
if (!config) {
|
if (!config) {
|
||||||
return {
|
return {
|
||||||
success: false,
|
ok: false as const,
|
||||||
error: "Configure Smartsheet first (save your API token and column mapping).",
|
error:
|
||||||
considered: 0,
|
"Configure Smartsheet first (save your API token and column mapping).",
|
||||||
newlyQueued: 0,
|
|
||||||
alreadyQueued: 0,
|
|
||||||
drained: { synced: 0, skipped: 0, failed: 0, remaining: 0 },
|
|
||||||
};
|
};
|
||||||
}
|
}
|
||||||
if (!config.syncEnabled) {
|
if (!config.syncEnabled) {
|
||||||
return {
|
return {
|
||||||
success: false,
|
ok: false as const,
|
||||||
error: "Smartsheet sync is disabled — enable it in the toggle above before backfilling.",
|
error:
|
||||||
considered: 0,
|
"Smartsheet sync is disabled — enable it in the toggle above before backfilling.",
|
||||||
newlyQueued: 0,
|
|
||||||
alreadyQueued: 0,
|
|
||||||
drained: { synced: 0, skipped: 0, failed: 0, remaining: 0 },
|
|
||||||
};
|
};
|
||||||
}
|
}
|
||||||
|
|
||||||
// 2. Find candidate entry IDs.
|
|
||||||
const entryWhere = opts.since
|
const entryWhere = opts.since
|
||||||
? and(
|
? and(
|
||||||
eq(waterLogEntries.brandId, brandId),
|
eq(waterLogEntries.brandId, brandId),
|
||||||
@@ -568,15 +567,13 @@ export async function enqueueSmartsheetBackfill(
|
|||||||
const entryIds = entryRows.map((r) => r.id);
|
const entryIds = entryRows.map((r) => r.id);
|
||||||
if (entryIds.length === 0) {
|
if (entryIds.length === 0) {
|
||||||
return {
|
return {
|
||||||
success: true,
|
ok: true as const,
|
||||||
considered: 0,
|
considered: 0,
|
||||||
newlyQueued: 0,
|
newlyQueued: 0,
|
||||||
alreadyQueued: 0,
|
alreadyQueued: 0,
|
||||||
drained: { synced: 0, skipped: 0, failed: 0, remaining: 0 },
|
|
||||||
};
|
};
|
||||||
}
|
}
|
||||||
|
|
||||||
// 3. Find which ones are already queued.
|
|
||||||
const existingRows = await db
|
const existingRows = await db
|
||||||
.select({ entryId: waterSmartsheetSyncQueue.entryId })
|
.select({ entryId: waterSmartsheetSyncQueue.entryId })
|
||||||
.from(waterSmartsheetSyncQueue)
|
.from(waterSmartsheetSyncQueue)
|
||||||
@@ -585,7 +582,6 @@ export async function enqueueSmartsheetBackfill(
|
|||||||
const toInsert = entryIds.filter((id) => !existingSet.has(id));
|
const toInsert = entryIds.filter((id) => !existingSet.has(id));
|
||||||
const alreadyQueued = entryIds.length - toInsert.length;
|
const alreadyQueued = entryIds.length - toInsert.length;
|
||||||
|
|
||||||
// 4. Insert in chunks (idempotent on the UNIQUE constraint).
|
|
||||||
let newlyQueued = 0;
|
let newlyQueued = 0;
|
||||||
for (let i = 0; i < toInsert.length; i += BACKFILL_INSERT_CHUNK) {
|
for (let i = 0; i < toInsert.length; i += BACKFILL_INSERT_CHUNK) {
|
||||||
const chunk = toInsert.slice(i, i + BACKFILL_INSERT_CHUNK);
|
const chunk = toInsert.slice(i, i + BACKFILL_INSERT_CHUNK);
|
||||||
@@ -607,22 +603,39 @@ export async function enqueueSmartsheetBackfill(
|
|||||||
newlyQueued += inserted.length;
|
newlyQueued += inserted.length;
|
||||||
}
|
}
|
||||||
|
|
||||||
// 5. Kick off a single drain pass for immediate feedback. Cron
|
return {
|
||||||
// picks up the rest over the next ticks.
|
ok: true as const,
|
||||||
let drained: EnqueueBackfillResult["drained"] = {
|
considered: entryIds.length,
|
||||||
synced: 0,
|
newlyQueued,
|
||||||
skipped: 0,
|
alreadyQueued,
|
||||||
failed: 0,
|
|
||||||
remaining: 0,
|
|
||||||
};
|
};
|
||||||
try {
|
});
|
||||||
const result = await drainSyncQueue(brandId, {
|
|
||||||
batchSize: BACKFILL_DRAIN_BATCH,
|
if (!phase1.ok) {
|
||||||
});
|
return {
|
||||||
const remainingRows = await db
|
success: false,
|
||||||
.select({
|
error: phase1.error,
|
||||||
id: waterSmartsheetSyncQueue.id,
|
considered: 0,
|
||||||
})
|
newlyQueued: 0,
|
||||||
|
alreadyQueued: 0,
|
||||||
|
drained: { synced: 0, skipped: 0, failed: 0, remaining: 0 },
|
||||||
|
};
|
||||||
|
}
|
||||||
|
|
||||||
|
// ── Phase 2: inline drain in a FRESH transaction (sees the inserts) ──
|
||||||
|
let drained: EnqueueBackfillResult["drained"] = {
|
||||||
|
synced: 0,
|
||||||
|
skipped: 0,
|
||||||
|
failed: 0,
|
||||||
|
remaining: 0,
|
||||||
|
};
|
||||||
|
try {
|
||||||
|
const result = await drainSyncQueue(brandId, {
|
||||||
|
batchSize: BACKFILL_DRAIN_BATCH,
|
||||||
|
});
|
||||||
|
const remainingRows = await withBrand(brandId, async (db) => {
|
||||||
|
return db
|
||||||
|
.select({ id: waterSmartsheetSyncQueue.id })
|
||||||
.from(waterSmartsheetSyncQueue)
|
.from(waterSmartsheetSyncQueue)
|
||||||
.where(
|
.where(
|
||||||
and(
|
and(
|
||||||
@@ -630,52 +643,53 @@ export async function enqueueSmartsheetBackfill(
|
|||||||
inArray(waterSmartsheetSyncQueue.status, ["pending", "failed"]),
|
inArray(waterSmartsheetSyncQueue.status, ["pending", "failed"]),
|
||||||
),
|
),
|
||||||
);
|
);
|
||||||
drained = {
|
|
||||||
synced: result.synced,
|
|
||||||
skipped: result.skipped,
|
|
||||||
failed: result.failed,
|
|
||||||
remaining: remainingRows.length,
|
|
||||||
};
|
|
||||||
} catch (err) {
|
|
||||||
// Don't fail the whole backfill if the inline drain errors —
|
|
||||||
// the cron will retry. Just surface to the result.
|
|
||||||
drained = {
|
|
||||||
synced: 0,
|
|
||||||
skipped: 0,
|
|
||||||
failed: 0,
|
|
||||||
remaining: -1,
|
|
||||||
// eslint-disable-next-line @typescript-eslint/no-explicit-any
|
|
||||||
...({ error: (err as Error).message } as any),
|
|
||||||
};
|
|
||||||
}
|
|
||||||
|
|
||||||
await logAuditEvent({
|
|
||||||
brandId,
|
|
||||||
actorId: auth.adminUser.user_id ?? null,
|
|
||||||
actorLabel: auth.adminUser.email ?? auth.adminUser.display_name ?? "admin",
|
|
||||||
action: "create",
|
|
||||||
entityType: "settings",
|
|
||||||
details: {
|
|
||||||
feature: "smartsheet",
|
|
||||||
event: "backfill",
|
|
||||||
considered: entryIds.length,
|
|
||||||
newlyQueued,
|
|
||||||
alreadyQueued,
|
|
||||||
drainedSynced: drained.synced,
|
|
||||||
drainedSkipped: drained.skipped,
|
|
||||||
drainedFailed: drained.failed,
|
|
||||||
since: opts.since?.toISOString() ?? null,
|
|
||||||
},
|
|
||||||
});
|
});
|
||||||
|
drained = {
|
||||||
return {
|
synced: result.synced,
|
||||||
success: true,
|
skipped: result.skipped,
|
||||||
considered: entryIds.length,
|
failed: result.failed,
|
||||||
newlyQueued,
|
remaining: remainingRows.length,
|
||||||
alreadyQueued,
|
|
||||||
drained,
|
|
||||||
};
|
};
|
||||||
|
} catch (err) {
|
||||||
|
// Don't fail the whole backfill if the inline drain errors —
|
||||||
|
// the cron will retry. Just surface to the result.
|
||||||
|
drained = {
|
||||||
|
synced: 0,
|
||||||
|
skipped: 0,
|
||||||
|
failed: 0,
|
||||||
|
remaining: -1,
|
||||||
|
// eslint-disable-next-line @typescript-eslint/no-explicit-any
|
||||||
|
...({ error: (err as Error).message } as any),
|
||||||
|
};
|
||||||
|
}
|
||||||
|
|
||||||
|
// ── Phase 3: audit log (logAuditEvent opens its own withBrand) ──
|
||||||
|
await logAuditEvent({
|
||||||
|
brandId,
|
||||||
|
actorId: auth.adminUser.user_id ?? null,
|
||||||
|
actorLabel: auth.adminUser.email ?? auth.adminUser.display_name ?? "admin",
|
||||||
|
action: "create",
|
||||||
|
entityType: "settings",
|
||||||
|
details: {
|
||||||
|
feature: "smartsheet",
|
||||||
|
event: "backfill",
|
||||||
|
considered: phase1.considered,
|
||||||
|
newlyQueued: phase1.newlyQueued,
|
||||||
|
alreadyQueued: phase1.alreadyQueued,
|
||||||
|
drainedSynced: drained.synced,
|
||||||
|
drainedSkipped: drained.skipped,
|
||||||
|
drainedFailed: drained.failed,
|
||||||
|
since: opts.since?.toISOString() ?? null,
|
||||||
|
},
|
||||||
});
|
});
|
||||||
|
|
||||||
|
return {
|
||||||
|
success: true,
|
||||||
|
considered: phase1.considered,
|
||||||
|
newlyQueued: phase1.newlyQueued,
|
||||||
|
alreadyQueued: phase1.alreadyQueued,
|
||||||
|
drained,
|
||||||
|
};
|
||||||
}
|
}
|
||||||
|
|
||||||
function validateColumnMapping(
|
function validateColumnMapping(
|
||||||
|
|||||||
@@ -21,6 +21,10 @@
|
|||||||
"path": "/api/square/process-queue",
|
"path": "/api/square/process-queue",
|
||||||
"schedule": "*/2 * * * *"
|
"schedule": "*/2 * * * *"
|
||||||
},
|
},
|
||||||
|
{
|
||||||
|
"path": "/api/water-log/smartsheet-sync",
|
||||||
|
"schedule": "*/15 * * * *"
|
||||||
|
},
|
||||||
{
|
{
|
||||||
"path": "/api/email-automation/abandoned-cart",
|
"path": "/api/email-automation/abandoned-cart",
|
||||||
"schedule": "0 */6 * * *"
|
"schedule": "0 */6 * * *"
|
||||||
|
|||||||
Reference in New Issue
Block a user