diff --git a/src/actions/water-log/smartsheet.ts b/src/actions/water-log/smartsheet.ts index 2989731..6fa72bb 100644 --- a/src/actions/water-log/smartsheet.ts +++ b/src/actions/water-log/smartsheet.ts @@ -15,9 +15,10 @@ * - Read actions return `maskedToken` ("••••abcd") and never the * plaintext. */ -import { desc, eq } from "drizzle-orm"; +import { desc, eq, and, gte, inArray } from "drizzle-orm"; import { withBrand } from "@/db/client"; import { + waterLogEntries, waterSmartsheetConfig, waterSmartsheetSyncLog, waterSmartsheetSyncQueue, @@ -29,7 +30,7 @@ import { } from "@/db/schema/water-log"; import { decryptToken, encryptToken, maskToken } from "@/lib/crypto"; import { extractSheetId, getSheetMeta, SmartsheetApiError } from "@/lib/smartsheet"; -import { syncEntryToSmartsheet } from "@/services/smartsheet-sync"; +import { syncEntryToSmartsheet, drainSyncQueue } from "@/services/smartsheet-sync"; import { getAdminUser } from "@/lib/admin-permissions"; import { logAuditEvent } from "@/lib/water-log-audit"; import { getSession } from "@/lib/auth"; @@ -452,6 +453,231 @@ export async function triggerSyncForEntry( // ── Helpers ──────────────────────────────────────────────────────────────── +/** + * Backfill: enqueue every existing water log entry for the brand so it + * gets pushed to Smartsheet. + * + * Use cases: + * - Customer configured Smartsheet after months/years of recording + * entries → they need historical data in the sheet. + * - Customer rotated their Smartsheet sheet (different ID) → they + * want a clean re-push without changing the per-entry hook. + * + * Safety: + * - Insert is `ON CONFLICT DO NOTHING` so re-runs are no-ops on + * already-queued entries. The UNIQUE(brand_id, entry_id) on the + * queue absorbs duplicates. + * - The sync engine dedups on the Smartsheet side by Entry ID + * (`findRowByColumn`), so even if we re-enqueue an old entry whose + * queue row was deleted, the sheet still won't gain duplicates. + * + * Performance / scope: + * - Inserts are chunked at 500 rows per round-trip. + * - After enqueueing, we kick off one inline drain pass + * (batchSize 200) so the admin gets immediate feedback. Any + * remaining rows are drained by the cron over time. + */ +export type EnqueueBackfillOptions = { + /** Only enqueue entries logged at-or-after this date. */ + since?: Date; +}; + +export type EnqueueBackfillResult = { + success: boolean; + error?: string; + /** Total entries in `water_log_entries` for the brand matching the filter. */ + considered: number; + /** Newly inserted rows in `water_smartsheet_sync_queue`. */ + newlyQueued: number; + /** Entries that were already in the queue (UNIQUE absorbed). */ + alreadyQueued: number; + /** + * Result of the inline drain pass that was kicked off after enqueue. + * `remaining` is the count of pending/failed queue rows that the + * next cron tick will pick up. + */ + drained: { + synced: number; + skipped: number; + failed: number; + remaining: number; + }; +}; + +const BACKFILL_INSERT_CHUNK = 500; +const BACKFILL_DRAIN_BATCH = 200; + +export async function enqueueSmartsheetBackfill( + brandId: string, + opts: EnqueueBackfillOptions = {}, +): Promise { + await getSession(); + const auth = await requireWaterAdminPermission(); + if (!auth.ok) { + const empty = { + considered: 0, + newlyQueued: 0, + alreadyQueued: 0, + drained: { synced: 0, skipped: 0, failed: 0, remaining: 0 }, + }; + return { success: false, error: auth.error, ...empty }; + } + + return withBrand(brandId, async (db) => { + // 1. Require active config. + const configRows = await db + .select({ + syncEnabled: waterSmartsheetConfig.syncEnabled, + }) + .from(waterSmartsheetConfig) + .where(eq(waterSmartsheetConfig.brandId, brandId)) + .limit(1); + const config = configRows[0]; + if (!config) { + return { + success: false, + error: "Configure Smartsheet first (save your API token and column mapping).", + considered: 0, + newlyQueued: 0, + alreadyQueued: 0, + drained: { synced: 0, skipped: 0, failed: 0, remaining: 0 }, + }; + } + if (!config.syncEnabled) { + return { + success: false, + error: "Smartsheet sync is disabled — enable it in the toggle above before backfilling.", + considered: 0, + newlyQueued: 0, + alreadyQueued: 0, + drained: { synced: 0, skipped: 0, failed: 0, remaining: 0 }, + }; + } + + // 2. Find candidate entry IDs. + const entryWhere = opts.since + ? and( + eq(waterLogEntries.brandId, brandId), + gte(waterLogEntries.loggedAt, opts.since), + ) + : eq(waterLogEntries.brandId, brandId); + const entryRows = await db + .select({ id: waterLogEntries.id }) + .from(waterLogEntries) + .where(entryWhere); + const entryIds = entryRows.map((r) => r.id); + if (entryIds.length === 0) { + return { + success: true, + considered: 0, + newlyQueued: 0, + alreadyQueued: 0, + drained: { synced: 0, skipped: 0, failed: 0, remaining: 0 }, + }; + } + + // 3. Find which ones are already queued. + const existingRows = await db + .select({ entryId: waterSmartsheetSyncQueue.entryId }) + .from(waterSmartsheetSyncQueue) + .where(eq(waterSmartsheetSyncQueue.brandId, brandId)); + const existingSet = new Set(existingRows.map((r) => r.entryId)); + const toInsert = entryIds.filter((id) => !existingSet.has(id)); + const alreadyQueued = entryIds.length - toInsert.length; + + // 4. Insert in chunks (idempotent on the UNIQUE constraint). + let newlyQueued = 0; + for (let i = 0; i < toInsert.length; i += BACKFILL_INSERT_CHUNK) { + const chunk = toInsert.slice(i, i + BACKFILL_INSERT_CHUNK); + const values = chunk.map((entryId) => ({ + brandId, + entryId, + status: "pending" as const, + })); + const inserted = await db + .insert(waterSmartsheetSyncQueue) + .values(values) + .onConflictDoNothing({ + target: [ + waterSmartsheetSyncQueue.brandId, + waterSmartsheetSyncQueue.entryId, + ], + }) + .returning({ entryId: waterSmartsheetSyncQueue.entryId }); + newlyQueued += inserted.length; + } + + // 5. Kick off a single drain pass for immediate feedback. Cron + // picks up the rest over the next ticks. + 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 db + .select({ + id: waterSmartsheetSyncQueue.id, + }) + .from(waterSmartsheetSyncQueue) + .where( + and( + eq(waterSmartsheetSyncQueue.brandId, brandId), + 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, + }, + }); + + return { + success: true, + considered: entryIds.length, + newlyQueued, + alreadyQueued, + drained, + }; + }); +} + function validateColumnMapping( m: unknown, ): { ok: true; value: SmartsheetColumnMapping } | { ok: false; error: string } { diff --git a/src/components/admin/water-log/SmartsheetIntegrationCard.tsx b/src/components/admin/water-log/SmartsheetIntegrationCard.tsx index 11311c8..367667d 100644 --- a/src/components/admin/water-log/SmartsheetIntegrationCard.tsx +++ b/src/components/admin/water-log/SmartsheetIntegrationCard.tsx @@ -32,6 +32,8 @@ import { saveSmartsheetConfig, testSmartsheetConnection, listSmartsheetSyncLog, + enqueueSmartsheetBackfill, + type EnqueueBackfillResult, type SmartsheetConfigResponse, type SmartsheetSyncLogEntry, } from "@/actions/water-log/smartsheet"; @@ -97,6 +99,12 @@ type State = { loadingLog: boolean; /** Whether the setup-instructions modal is open. */ setupModalOpen: boolean; + /** Backfill UI state. */ + backfill: { + running: boolean; + range: "all" | "30d" | "7d"; + result: EnqueueBackfillResult | null; + }; }; type Action = @@ -118,7 +126,10 @@ type Action = | { type: "CLEAR_MESSAGE" } | { type: "REFRESH_LOG"; log: SmartsheetSyncLogEntry[] } | { type: "OPEN_SETUP_MODAL" } - | { type: "CLOSE_SETUP_MODAL" }; + | { type: "CLOSE_SETUP_MODAL" } + | { type: "BACKFILL_START" } + | { type: "BACKFILL_RANGE"; value: "all" | "30d" | "7d" } + | { type: "BACKFILL_DONE"; result: EnqueueBackfillResult }; const EMPTY_MAPPING: SmartsheetColumnMapping = { entry_id: "", @@ -146,6 +157,7 @@ const initialState: State = { recentLog: [], loadingLog: false, setupModalOpen: false, + backfill: { running: false, range: "all", result: null }, }; function reducer(state: State, action: Action): State { @@ -227,6 +239,25 @@ function reducer(state: State, action: Action): State { return { ...state, setupModalOpen: true }; case "CLOSE_SETUP_MODAL": return { ...state, setupModalOpen: false }; + case "BACKFILL_START": + return { + ...state, + backfill: { ...state.backfill, running: true }, + }; + case "BACKFILL_RANGE": + return { + ...state, + backfill: { ...state.backfill, range: action.value }, + }; + case "BACKFILL_DONE": + return { + ...state, + backfill: { + ...state.backfill, + running: false, + result: action.result, + }, + }; default: return state; } @@ -313,6 +344,50 @@ export default function SmartsheetIntegrationCard({ } } + async function handleBackfill() { + if (state.backfill.running) return; + if (!state.savedConfig?.configured || !state.enabled) { + // Surface as a notice rather than call into the action just to fail. + dispatch({ + type: "BACKFILL_DONE", + result: { + success: false, + error: "Save and enable your Smartsheet config first.", + considered: 0, + newlyQueued: 0, + alreadyQueued: 0, + drained: { synced: 0, skipped: 0, failed: 0, remaining: 0 }, + }, + }); + return; + } + const rangeLabel = + state.backfill.range === "30d" + ? "the last 30 days" + : state.backfill.range === "7d" + ? "the last 7 days" + : "all time"; + if ( + !window.confirm( + `Sync ${rangeLabel} of water log entries to Smartsheet? This is safe to re-run — already-pushed entries are deduped by Entry ID.`, + ) + ) { + return; + } + dispatch({ type: "BACKFILL_START" }); + const since = + state.backfill.range === "30d" + ? new Date(Date.now() - 30 * 86_400_000) + : state.backfill.range === "7d" + ? new Date(Date.now() - 7 * 86_400_000) + : undefined; + const result = await enqueueSmartsheetBackfill(brandId, { since }); + dispatch({ type: "BACKFILL_DONE", result }); + // Refresh the log so the drain results show up immediately. + const log = await listSmartsheetSyncLog(brandId, 10); + dispatch({ type: "REFRESH_LOG", log }); + } + if (state.loading) { return ( @@ -570,6 +645,8 @@ export default function SmartsheetIntegrationCard({ + dispatch({ type: "BACKFILL_RANGE", value: v })} /> + void; + onRangeChange: (value: "all" | "30d" | "7d") => void; +}) { + const result = state.result; + // Success block — counts from the inline drain that ran after enqueue. + const showResult = result && result.success; + const showError = result && !result.success; + + return ( + +
+
+ + + +
+ + {state.running && ( +

+ Counting entries, enqueueing, and draining one batch inline. Any + remaining entries will be drained by the cron (every 15 minutes). +

+ )} + + {showError && ( +
+

✗ {result.error}

+
+ )} + + {showResult && result && ( +
+

+ ✓ Found {result.considered}{" "} + {result.considered === 1 ? "entry" : "entries"} matching the + range + {result.considered > 0 && ( + <> + {" "} + ({result.newlyQueued} new, {result.alreadyQueued} already + queued). + + )} + . +

+ {(result.drained.synced + + result.drained.skipped + + result.drained.failed > + 0 || result.drained.remaining > 0) && ( +

+ Inline drain: {result.drained.synced} synced /{" "} + {result.drained.skipped} skipped / {result.drained.failed}{" "} + failed. + {result.drained.remaining > 0 && ( + <> + {" "} + + {result.drained.remaining} still queued + {" "} + — they'll drain on the next cron tick (within 15 + minutes). + + )} + {result.drained.remaining === 0 && + result.considered > 0 && ( + <> All entries are now in Smartsheet. + )} +

+ )} +
+ )} +
+ + ); +} + // ── Setup instructions modal ─────────────────────────────────────────────── const FIELD_SCHEMA: Array<{