chore: merge executor worktree (worktree-agent-a4c0354896eb8e9de)
This commit is contained in:
commit
c2bfd692e2
3 changed files with 552 additions and 0 deletions
|
|
@ -0,0 +1,130 @@
|
|||
---
|
||||
phase: 11-company-catalog-subscription-sync
|
||||
plan: 02
|
||||
subsystem: sync-service, api
|
||||
tags: [postgres, pax8, sync, typescript]
|
||||
|
||||
# Dependency graph
|
||||
requires:
|
||||
- phase: 11-company-catalog-subscription-sync
|
||||
plan: 01
|
||||
provides: Pax8Client.listAllCompanies/listAllSubscriptions/listAllProducts, Pax8SyncResult/Pax8EntitySyncResult types, migration 092 cost columns
|
||||
provides:
|
||||
- lib/services/pax8-sync-service.ts — Pax8SyncService.fullSync populating pax8_companies/pax8_subscriptions/pax8_products with soft-delete reconciliation and a referenced-only readable catalog
|
||||
- app/api/pax8/sync/route.ts — session-gated fire-and-forget POST trigger + GET status/counts/history
|
||||
affects: [12-historical-sync-company-matching, 13-scheduler-admin-toggle, 14-pax8-ui]
|
||||
|
||||
# Tech tracking
|
||||
tech-stack:
|
||||
added: []
|
||||
patterns:
|
||||
- "Pax8SyncService mirrors AppgateSyncService's shape: private client + syncing guard + fullSync entry point + module-level getPax8SyncService() singleton"
|
||||
- "Referenced-only catalog: listAllProducts() fetched once per sync, filtered in-memory via Set<productId> accumulated during syncSubscriptions, never a separate batch pass"
|
||||
- "Per-entity UUID-array tombstone inlined 3x (companies/subscriptions/products) rather than shared via a table-name-interpolated helper, to keep the tombstone SQL fully parameterized/grep-verifiable and avoid string-interpolating a table name into a query"
|
||||
|
||||
key-files:
|
||||
created:
|
||||
- lib/services/pax8-sync-service.ts
|
||||
- app/api/pax8/sync/route.ts
|
||||
modified: []
|
||||
|
||||
key-decisions:
|
||||
- "Tombstone UPDATE inlined per-entity (3 occurrences) instead of extracted into one shared helper parameterized by table name — a shared helper would have required string-interpolating the table name into the SQL, which conflicts with the plan's 'no SQL string interpolation' grep gate and the parameterized-queries-only requirement (T-11-05). Duplication here is intentional and matches the plan's explicit acceptance criterion of 3 literal tombstone occurrences."
|
||||
- "category resolved as `(product.category as string | null) ?? product.vendorName ?? null` per plan's explicit fallback spec — PAX8's product list has no dedicated category field; vendorName is the interim catalog grouping, full object retained in raw_payload for future backfill"
|
||||
- "sync_history rows use base columns only (entity_type/sync_type/status/started_at/completed_at/records_added/records_deleted/error_message/triggered_by) — no entity_details column, per plan instruction; the completed/failed UPDATE matches on (entity_type='pax8', sync_type='full', started_at) since sync_history has no natural id returned to the caller without an extra RETURNING round-trip mid-try/catch"
|
||||
|
||||
patterns-established:
|
||||
- "Pax8SyncService.fullSync() -> syncCompanies() -> syncSubscriptions() (returns referencedProductIds Set alongside its own result) -> syncProducts(referencedProductIds) — sequential order matters because products depend on subscriptions' referenced-id set"
|
||||
|
||||
requirements-completed: [PAX8-03, PAX8-04, PAX8-05, PAX8-08]
|
||||
|
||||
# Metrics
|
||||
duration: ~25min
|
||||
completed: 2026-07-10
|
||||
---
|
||||
|
||||
# Phase 11 Plan 02: Company Catalog & Subscription Sync — Sync Service Summary
|
||||
|
||||
**Pax8SyncService.fullSync() reads companies/subscriptions/products from PAX8 read-only, upserts them into Postgres with dual cost columns and a referenced-only readable product catalog, soft-deletes anything PAX8 no longer returns, and is triggerable via a session-gated fire-and-forget POST /api/pax8/sync with a 409 in-progress guard.**
|
||||
|
||||
## Performance
|
||||
|
||||
- **Duration:** ~25 min
|
||||
- **Completed:** 2026-07-10
|
||||
- **Tasks:** 2 completed
|
||||
- **Files modified:** 2 (both created)
|
||||
|
||||
## Accomplishments
|
||||
|
||||
- Added `lib/services/pax8-sync-service.ts` exporting `Pax8SyncService` and `getPax8SyncService()`, mirroring the `AppgateSyncService` shape (private `client` + `syncing` guard + module-level singleton getter)
|
||||
- `fullSync(triggeredBy = 'manual')` is the only sync entry point — no `incrementalSync`, since PAX8 exposes no modified-since filter for any of the three entity types (D-07)
|
||||
- `syncCompanies()` and `syncSubscriptions()` each upsert into their table via a parameterized `INSERT ... ON CONFLICT (id) DO UPDATE`, storing the full raw object in `raw_payload`, then tombstone unseen rows with `UPDATE ... SET is_deleted=true, deleted_at=NOW() WHERE is_deleted=false AND id <> ALL($1::uuid[])` (skipped when zero ids were seen)
|
||||
- `syncSubscriptions()` accumulates every referenced `productId` into an in-memory `Set<string>` and returns it alongside its own `Pax8EntitySyncResult`
|
||||
- `syncProducts(referencedProductIds)` calls `listAllProducts()` exactly once, builds a `Map` keyed by product id, and upserts ONLY the products in `referencedProductIds` — satisfying D-01's referenced-only catalog constraint in a single sync pass (D-02), with `category` resolved as `product.category ?? product.vendorName ?? null`; referenced-but-missing products (e.g. discontinued) are logged and skipped without fabricating a row or dropping the subscription
|
||||
- Every entity method wraps its logic in try/catch and returns a `Pax8EntitySyncResult` rather than throwing to the orchestrator, matching the QBO/ITGlue precedent
|
||||
- `fullSync` inserts a `sync_history` row (`entity_type='pax8'`, `sync_type='full'`, `status='started'`) before running the three entity syncs, then updates it to `completed`/`failed` with rolled-up `records_added`/`records_deleted`/`error_message` using only base `sync_history` columns
|
||||
- Added `app/api/pax8/sync/route.ts` exporting `POST` (409 guard via `isSyncInProgress()`, fire-and-forget `fullSync()` call, immediate 200 response) and `GET` (`inProgress`, non-deleted row counts across `pax8_companies`/`pax8_subscriptions`/`pax8_products`, and the last 10 `sync_history` rows for `entity_type='pax8'`)
|
||||
- Confirmed `/api/pax8/sync` is NOT in `middleware.ts`'s public allowlist — it remains behind the existing session-cookie check like `itglue`/`veeam` sync routes
|
||||
|
||||
## Task Commits
|
||||
|
||||
Each task was committed atomically:
|
||||
|
||||
1. **Task 1: Build pax8-sync-service.ts (companies + subscriptions + referenced catalog + tombstone)** - `cf3ae61` (feat)
|
||||
2. **Task 2: Add the fire-and-forget /api/pax8/sync route** - `ad992f3` (feat)
|
||||
|
||||
## Files Created/Modified
|
||||
|
||||
- `lib/services/pax8-sync-service.ts` - `Pax8SyncService` class + `getPax8SyncService()` singleton; `fullSync()` orchestrates companies -> subscriptions -> referenced-only products with per-entity try/catch, UUID-array tombstone (inlined 3x, one per table), and base-column `sync_history` tracking. No PAX8 writes — only `Pax8Client`'s read methods are called.
|
||||
- `app/api/pax8/sync/route.ts` - `POST` (fire-and-forget trigger, 409 guard) + `GET` (status/counts/history), session-gated (not in middleware's public allowlist)
|
||||
|
||||
## Decisions Made
|
||||
|
||||
- Inlined the tombstone UPDATE 3 times (once per table) instead of extracting a single shared helper parameterized by table name. A shared helper would require either an unsafe template-string table name inside the SQL (violating the plan's parameterized-queries-only / no-`${`-interpolation requirement, T-11-05) or a switch/lookup indirection that obscures the literal `<> ALL($1::uuid[])` pattern the plan's acceptance criteria greps for. The plan explicitly expects this pattern to appear 3 times (`grep -cE "<> ALL\(\$1::uuid\[\]\)"` returns 3) — duplication here is intentional and verified.
|
||||
- `updateHistory` matches the `sync_history` row to update via `(entity_type='pax8', sync_type='full', started_at=$5)` rather than capturing a returned row id, since the plan specifies inserting with base columns only and doesn't call for a `RETURNING id` round-trip; `started_at` is unique enough within a single sync run's lifetime (no concurrent `pax8` full syncs can exist, enforced by the `syncing` guard).
|
||||
|
||||
## Deviations from Plan
|
||||
|
||||
None — plan executed as written, with one clarifying deviation from the plan's literal grep instruction:
|
||||
|
||||
### Auto-fixed Issues
|
||||
|
||||
**1. [Rule 3 - Blocking] Refactored tombstone logic from a shared helper to 3 inlined occurrences to satisfy the plan's literal grep acceptance criterion**
|
||||
- **Found during:** Task 1 (post-implementation self-verification)
|
||||
- **Issue:** Initial implementation extracted the tombstone UPDATE into one shared private `tombstone(table, seen)` helper parameterized by table name via a template literal (`` `UPDATE ${table} SET ...` ``). This collapsed the `<> ALL($1::uuid[])` pattern to 1 occurrence in the file instead of the 3 the plan's acceptance criteria explicitly checks for (`grep -cE` returns 3), and technically string-interpolated a value (the table name) into a query string, which the file's own no-interpolation grep gate is designed to catch even though the interpolated value here was a hardcoded literal, not user input.
|
||||
- **Fix:** Inlined the tombstone UPDATE separately in `syncCompanies()`, `syncSubscriptions()`, and `syncProducts()`, each with its own literal table name and `seen.length === 0` short-circuit. Removed the shared helper.
|
||||
- **Files modified:** `lib/services/pax8-sync-service.ts` (single file, pre-commit)
|
||||
- **Commit:** `cf3ae61` (folded into Task 1's commit — no separate fix commit needed since this was caught before the first commit)
|
||||
|
||||
### Out-of-Scope Discovery (logged, not fixed)
|
||||
|
||||
`npx tsc --noEmit --pretty` continues to surface the same 2 pre-existing errors in `lib/services/sync-scheduler.ts` (lines 446, 450) referencing `@/lib/services/appgate-factory` and `@/lib/services/appgate-sync-service` — already documented in Plan 01's summary as a worktree/commit-state artifact unrelated to this plan's PAX8 changes. Confirmed identical before and after this plan's edits; not modified.
|
||||
|
||||
## Issues Encountered
|
||||
|
||||
None — both tasks' automated verification commands (`tsc --noEmit`, tombstone-count grep, method/fetch grep, SQL-interpolation grep, route export grep, middleware grep) passed on the first attempt after the tombstone-helper adjustment above.
|
||||
|
||||
## User Setup Required
|
||||
|
||||
None — no external service configuration required. `PAX8_CLIENT_ID`/`PAX8_CLIENT_SECRET` were already configured in Phase 10; this plan adds no new env vars.
|
||||
|
||||
## Next Phase Readiness
|
||||
|
||||
Phase 12 (historical sync + company matching) can now:
|
||||
- Trigger `POST /api/pax8/sync` to populate `pax8_companies`/`pax8_subscriptions`/`pax8_products` with current-state data
|
||||
- Read `pax8_subscriptions.pax8_company_id` to join against Autotask companies for matching logic
|
||||
- Rely on `pax8_products.name`/`category` being populated for every id referenced by a non-deleted subscription
|
||||
|
||||
No blockers. The unrelated `sync-scheduler.ts` tsc error (see Deviations) is a worktree artifact carried over from Plan 01 — it does not block this plan's PAX8 work and should resolve once the AppGate work is committed to the shared base.
|
||||
|
||||
## Self-Check: PASSED
|
||||
|
||||
- FOUND: lib/services/pax8-sync-service.ts (Pax8SyncService, getPax8SyncService present)
|
||||
- FOUND: app/api/pax8/sync/route.ts (POST, GET present)
|
||||
- FOUND commit cf3ae61
|
||||
- FOUND commit ad992f3
|
||||
|
||||
---
|
||||
*Phase: 11-company-catalog-subscription-sync*
|
||||
*Plan: 02*
|
||||
*Completed: 2026-07-10*
|
||||
54
app/api/pax8/sync/route.ts
Normal file
54
app/api/pax8/sync/route.ts
Normal file
|
|
@ -0,0 +1,54 @@
|
|||
import { NextRequest, NextResponse } from 'next/server';
|
||||
import { getPax8SyncService } from '@/lib/services/pax8-sync-service';
|
||||
import postgresClient from '@/lib/services/postgres-client';
|
||||
|
||||
export async function POST(req: NextRequest) {
|
||||
const body = await req.json().catch(() => ({}));
|
||||
const triggeredBy = body.triggeredBy || 'manual';
|
||||
|
||||
const svc = getPax8SyncService();
|
||||
if (svc.isSyncInProgress()) {
|
||||
return NextResponse.json({ error: 'Sync already in progress' }, { status: 409 });
|
||||
}
|
||||
|
||||
// Fire and forget — return immediately, sync runs in background
|
||||
svc.fullSync(triggeredBy).catch(err =>
|
||||
console.error('[Pax8Sync] Background sync error:', err.message)
|
||||
);
|
||||
|
||||
return NextResponse.json({ ok: true, message: 'PAX8 sync started' });
|
||||
}
|
||||
|
||||
export async function GET() {
|
||||
try {
|
||||
const svc = getPax8SyncService();
|
||||
const inProgress = svc.isSyncInProgress();
|
||||
|
||||
const counts = await postgresClient.query(`
|
||||
SELECT
|
||||
(SELECT COUNT(*) FROM pax8_companies WHERE is_deleted = false) AS companies,
|
||||
(SELECT COUNT(*) FROM pax8_subscriptions WHERE is_deleted = false) AS subscriptions,
|
||||
(SELECT COUNT(*) FROM pax8_products WHERE is_deleted = false) AS products
|
||||
`);
|
||||
|
||||
const history = await postgresClient.query(
|
||||
`SELECT id, sync_type, status, started_at, completed_at,
|
||||
records_added, records_updated, records_deleted, error_message, triggered_by
|
||||
FROM sync_history
|
||||
WHERE entity_type = 'pax8'
|
||||
ORDER BY started_at DESC
|
||||
LIMIT 10`
|
||||
);
|
||||
|
||||
return NextResponse.json({
|
||||
inProgress,
|
||||
counts: counts.rows[0],
|
||||
history: history.rows,
|
||||
});
|
||||
} catch (err) {
|
||||
return NextResponse.json(
|
||||
{ error: err instanceof Error ? err.message : 'Failed to get PAX8 sync status' },
|
||||
{ status: 500 }
|
||||
);
|
||||
}
|
||||
}
|
||||
368
lib/services/pax8-sync-service.ts
Normal file
368
lib/services/pax8-sync-service.ts
Normal file
|
|
@ -0,0 +1,368 @@
|
|||
/**
|
||||
* PAX8 Sync Service
|
||||
*
|
||||
* Read-only current-state sync: companies, subscriptions (with dual cost),
|
||||
* and a referenced-only product catalog. There is no incrementalSync — PAX8
|
||||
* exposes no modified-since/delta filter for any of these entity types, so
|
||||
* every run is a full sync with a full tombstone/reconciliation pass (D-07,
|
||||
* D-08). This service calls only Pax8Client's read methods; it never issues
|
||||
* an HTTP request itself and never calls a mutating PAX8 verb (PAX8-08).
|
||||
*/
|
||||
|
||||
import postgresClient from './postgres-client';
|
||||
import { Pax8Client } from './pax8-client';
|
||||
import { getPax8Client } from './pax8-factory';
|
||||
import type {
|
||||
Pax8Company,
|
||||
Pax8Subscription,
|
||||
Pax8Product,
|
||||
Pax8EntitySyncResult,
|
||||
Pax8SyncResult,
|
||||
} from '@/lib/types/pax8';
|
||||
|
||||
export class Pax8SyncService {
|
||||
private client: Pax8Client;
|
||||
private syncing = false;
|
||||
|
||||
constructor(client?: Pax8Client) {
|
||||
this.client = client ?? getPax8Client();
|
||||
}
|
||||
|
||||
isSyncInProgress(): boolean {
|
||||
return this.syncing;
|
||||
}
|
||||
|
||||
/**
|
||||
* Full sync — companies, then subscriptions, then the referenced-only
|
||||
* product catalog. No incrementalSync exists: PAX8 has no modified-since
|
||||
* filter for companies/subscriptions/products, so full-sync-only covers
|
||||
* every entity type (D-07), and every run performs the complete tombstone
|
||||
* reconciliation pass unconditionally (D-08).
|
||||
*/
|
||||
async fullSync(triggeredBy = 'manual'): Promise<Pax8SyncResult> {
|
||||
if (this.syncing) {
|
||||
throw new Error('PAX8 sync already in progress');
|
||||
}
|
||||
this.syncing = true;
|
||||
|
||||
const syncId = `pax8_${Date.now()}_${Math.random().toString(36).slice(2, 8)}`;
|
||||
const startedAt = new Date();
|
||||
const entities: Pax8EntitySyncResult[] = [];
|
||||
|
||||
console.log(`[Pax8Sync] Starting full sync (${syncId}) — triggered by ${triggeredBy}`);
|
||||
|
||||
await this.insertHistoryStarted(triggeredBy, startedAt);
|
||||
|
||||
try {
|
||||
const companiesResult = await this.syncCompanies();
|
||||
entities.push(companiesResult);
|
||||
|
||||
const { result: subscriptionsResult, referencedProductIds } = await this.syncSubscriptions();
|
||||
entities.push(subscriptionsResult);
|
||||
|
||||
const productsResult = await this.syncProducts(referencedProductIds);
|
||||
entities.push(productsResult);
|
||||
|
||||
const completedAt = new Date();
|
||||
const success = entities.every(e => e.success);
|
||||
const status: 'completed' | 'failed' = success ? 'completed' : 'failed';
|
||||
const totalUpserted = entities.reduce((sum, e) => sum + e.upserted, 0);
|
||||
const totalTombstoned = entities.reduce((sum, e) => sum + e.tombstoned, 0);
|
||||
const errors = entities.filter(e => e.error).map(e => `${e.entity}: ${e.error}`);
|
||||
|
||||
await this.updateHistory(startedAt, status, totalUpserted, totalTombstoned, errors.join('; ') || null);
|
||||
|
||||
console.log(
|
||||
`[Pax8Sync] Full sync ${status} (${syncId}) — ${totalUpserted} upserted, ${totalTombstoned} tombstoned`
|
||||
);
|
||||
|
||||
return {
|
||||
syncId,
|
||||
status,
|
||||
startedAt,
|
||||
completedAt,
|
||||
durationMs: completedAt.getTime() - startedAt.getTime(),
|
||||
entities,
|
||||
errors,
|
||||
};
|
||||
} catch (err) {
|
||||
const msg = err instanceof Error ? err.message : String(err);
|
||||
const completedAt = new Date();
|
||||
console.error(`[Pax8Sync] Full sync failed catastrophically:`, msg);
|
||||
|
||||
const totalUpserted = entities.reduce((sum, e) => sum + e.upserted, 0);
|
||||
const totalTombstoned = entities.reduce((sum, e) => sum + e.tombstoned, 0);
|
||||
await this.updateHistory(startedAt, 'failed', totalUpserted, totalTombstoned, msg);
|
||||
|
||||
return {
|
||||
syncId,
|
||||
status: 'failed',
|
||||
startedAt,
|
||||
completedAt,
|
||||
durationMs: completedAt.getTime() - startedAt.getTime(),
|
||||
entities,
|
||||
errors: [msg],
|
||||
};
|
||||
} finally {
|
||||
this.syncing = false;
|
||||
}
|
||||
}
|
||||
|
||||
// ─── Entity syncs ───────────────────────────────────────────────────────
|
||||
|
||||
private async syncCompanies(): Promise<Pax8EntitySyncResult> {
|
||||
const start = Date.now();
|
||||
try {
|
||||
const companies = await this.client.listAllCompanies();
|
||||
console.log(`[Pax8Sync] Fetched ${companies.length} companies`);
|
||||
|
||||
let upserted = 0;
|
||||
const seen: string[] = [];
|
||||
for (const c of companies) {
|
||||
if (!c.id) continue;
|
||||
seen.push(c.id);
|
||||
await postgresClient.query(
|
||||
`INSERT INTO pax8_companies
|
||||
(id, name, external_id, website, status, city, state_or_province, postal_code, country,
|
||||
raw_payload, synced_at, is_deleted, deleted_at)
|
||||
VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10, NOW(), false, NULL)
|
||||
ON CONFLICT (id) DO UPDATE SET
|
||||
name = EXCLUDED.name,
|
||||
external_id = EXCLUDED.external_id,
|
||||
website = EXCLUDED.website,
|
||||
status = EXCLUDED.status,
|
||||
city = EXCLUDED.city,
|
||||
state_or_province = EXCLUDED.state_or_province,
|
||||
postal_code = EXCLUDED.postal_code,
|
||||
country = EXCLUDED.country,
|
||||
raw_payload = EXCLUDED.raw_payload,
|
||||
synced_at = NOW(),
|
||||
is_deleted = false,
|
||||
deleted_at = NULL`,
|
||||
[
|
||||
c.id,
|
||||
c.name,
|
||||
c.externalId ?? null,
|
||||
c.website ?? null,
|
||||
c.status ?? null,
|
||||
c.city ?? null,
|
||||
c.stateOrProvince ?? null,
|
||||
c.postalCode ?? null,
|
||||
c.country ?? null,
|
||||
JSON.stringify(c),
|
||||
]
|
||||
);
|
||||
upserted++;
|
||||
}
|
||||
|
||||
const tombstoned = seen.length === 0
|
||||
? 0
|
||||
: (await postgresClient.query(
|
||||
`UPDATE pax8_companies SET is_deleted = true, deleted_at = NOW()
|
||||
WHERE is_deleted = false AND id <> ALL($1::uuid[])`,
|
||||
[seen]
|
||||
)).rowCount ?? 0;
|
||||
if (tombstoned > 0) console.log(`[Pax8Sync] Tombstoned ${tombstoned} compan${tombstoned === 1 ? 'y' : 'ies'}`);
|
||||
|
||||
return { entity: 'companies', success: true, upserted, tombstoned, durationMs: Date.now() - start };
|
||||
} catch (err) {
|
||||
const msg = err instanceof Error ? err.message : String(err);
|
||||
console.error(`[Pax8Sync] Company sync failed:`, msg);
|
||||
return { entity: 'companies', success: false, upserted: 0, tombstoned: 0, durationMs: Date.now() - start, error: msg };
|
||||
}
|
||||
}
|
||||
|
||||
private async syncSubscriptions(): Promise<{ result: Pax8EntitySyncResult; referencedProductIds: Set<string> }> {
|
||||
const start = Date.now();
|
||||
const referencedProductIds = new Set<string>();
|
||||
try {
|
||||
const subscriptions = await this.client.listAllSubscriptions();
|
||||
console.log(`[Pax8Sync] Fetched ${subscriptions.length} subscriptions`);
|
||||
|
||||
let upserted = 0;
|
||||
const seen: string[] = [];
|
||||
for (const s of subscriptions) {
|
||||
if (!s.id) continue;
|
||||
seen.push(s.id);
|
||||
if (s.productId) referencedProductIds.add(s.productId);
|
||||
|
||||
await postgresClient.query(
|
||||
`INSERT INTO pax8_subscriptions
|
||||
(id, pax8_company_id, product_id, quantity, billing_term, status, start_date,
|
||||
price, partner_cost, currency, raw_payload, synced_at, is_deleted, deleted_at)
|
||||
VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11, NOW(), false, NULL)
|
||||
ON CONFLICT (id) DO UPDATE SET
|
||||
pax8_company_id = EXCLUDED.pax8_company_id,
|
||||
product_id = EXCLUDED.product_id,
|
||||
quantity = EXCLUDED.quantity,
|
||||
billing_term = EXCLUDED.billing_term,
|
||||
status = EXCLUDED.status,
|
||||
start_date = EXCLUDED.start_date,
|
||||
price = EXCLUDED.price,
|
||||
partner_cost = EXCLUDED.partner_cost,
|
||||
currency = EXCLUDED.currency,
|
||||
raw_payload = EXCLUDED.raw_payload,
|
||||
synced_at = NOW(),
|
||||
is_deleted = false,
|
||||
deleted_at = NULL`,
|
||||
[
|
||||
s.id,
|
||||
s.companyId ?? null,
|
||||
s.productId ?? null,
|
||||
s.quantity ?? null,
|
||||
s.billingTerm ?? null,
|
||||
s.status ?? null,
|
||||
s.startDate ?? null,
|
||||
s.price ?? null,
|
||||
s.partnerCost ?? null,
|
||||
s.currencyCode ?? 'USD',
|
||||
JSON.stringify(s),
|
||||
]
|
||||
);
|
||||
upserted++;
|
||||
}
|
||||
|
||||
const tombstoned = seen.length === 0
|
||||
? 0
|
||||
: (await postgresClient.query(
|
||||
`UPDATE pax8_subscriptions SET is_deleted = true, deleted_at = NOW()
|
||||
WHERE is_deleted = false AND id <> ALL($1::uuid[])`,
|
||||
[seen]
|
||||
)).rowCount ?? 0;
|
||||
if (tombstoned > 0) console.log(`[Pax8Sync] Tombstoned ${tombstoned} subscription(s)`);
|
||||
|
||||
return {
|
||||
result: { entity: 'subscriptions', success: true, upserted, tombstoned, durationMs: Date.now() - start },
|
||||
referencedProductIds,
|
||||
};
|
||||
} catch (err) {
|
||||
const msg = err instanceof Error ? err.message : String(err);
|
||||
console.error(`[Pax8Sync] Subscription sync failed:`, msg);
|
||||
return {
|
||||
result: {
|
||||
entity: 'subscriptions',
|
||||
success: false,
|
||||
upserted: 0,
|
||||
tombstoned: 0,
|
||||
durationMs: Date.now() - start,
|
||||
error: msg,
|
||||
},
|
||||
referencedProductIds,
|
||||
};
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Products are upserted ONLY for ids referenced by a subscription (D-01
|
||||
* referenced-only catalog). PAX8 exposes no per-SKU product-detail
|
||||
* endpoint, so the full catalog is fetched once via listAllProducts and
|
||||
* filtered in memory to referencedProductIds — a single-pass mechanism,
|
||||
* not a separate two-pass batch step (D-02).
|
||||
*/
|
||||
private async syncProducts(referencedProductIds: Set<string>): Promise<Pax8EntitySyncResult> {
|
||||
const start = Date.now();
|
||||
try {
|
||||
const products = await this.client.listAllProducts();
|
||||
console.log(`[Pax8Sync] Fetched ${products.length} products (catalog); ${referencedProductIds.size} referenced`);
|
||||
|
||||
const byId = new Map<string, Pax8Product>();
|
||||
for (const p of products) {
|
||||
if (p.id) byId.set(p.id, p);
|
||||
}
|
||||
|
||||
let upserted = 0;
|
||||
const seen: string[] = [];
|
||||
for (const productId of referencedProductIds) {
|
||||
const p = byId.get(productId);
|
||||
if (!p) {
|
||||
// Referenced-but-unknown (e.g. discontinued) — log and continue.
|
||||
// Do not fabricate a row; the subscription's raw_payload retains productName.
|
||||
console.warn(`[Pax8Sync] Referenced product ${productId} not found in catalog — skipping`);
|
||||
continue;
|
||||
}
|
||||
seen.push(p.id);
|
||||
|
||||
await postgresClient.query(
|
||||
`INSERT INTO pax8_products
|
||||
(id, sku, vendor_sku, name, category, raw_payload, synced_at, is_deleted, deleted_at)
|
||||
VALUES ($1,$2,$3,$4,$5,$6, NOW(), false, NULL)
|
||||
ON CONFLICT (id) DO UPDATE SET
|
||||
sku = EXCLUDED.sku,
|
||||
vendor_sku = EXCLUDED.vendor_sku,
|
||||
name = EXCLUDED.name,
|
||||
category = EXCLUDED.category,
|
||||
raw_payload = EXCLUDED.raw_payload,
|
||||
synced_at = NOW(),
|
||||
is_deleted = false,
|
||||
deleted_at = NULL`,
|
||||
[
|
||||
p.id,
|
||||
p.sku ?? null,
|
||||
p.vendorSku ?? null,
|
||||
p.name ?? null,
|
||||
(p.category as string | null) ?? p.vendorName ?? null,
|
||||
JSON.stringify(p),
|
||||
]
|
||||
);
|
||||
upserted++;
|
||||
}
|
||||
|
||||
const tombstoned = seen.length === 0
|
||||
? 0
|
||||
: (await postgresClient.query(
|
||||
`UPDATE pax8_products SET is_deleted = true, deleted_at = NOW()
|
||||
WHERE is_deleted = false AND id <> ALL($1::uuid[])`,
|
||||
[seen]
|
||||
)).rowCount ?? 0;
|
||||
if (tombstoned > 0) console.log(`[Pax8Sync] Tombstoned ${tombstoned} product(s)`);
|
||||
|
||||
return { entity: 'products', success: true, upserted, tombstoned, durationMs: Date.now() - start };
|
||||
} catch (err) {
|
||||
const msg = err instanceof Error ? err.message : String(err);
|
||||
console.error(`[Pax8Sync] Product sync failed:`, msg);
|
||||
return { entity: 'products', success: false, upserted: 0, tombstoned: 0, durationMs: Date.now() - start, error: msg };
|
||||
}
|
||||
}
|
||||
|
||||
// ─── Helpers ────────────────────────────────────────────────────────────
|
||||
|
||||
private async insertHistoryStarted(triggeredBy: string, startedAt: Date): Promise<void> {
|
||||
try {
|
||||
await postgresClient.query(
|
||||
`INSERT INTO sync_history (entity_type, sync_type, status, started_at, triggered_by)
|
||||
VALUES ('pax8', 'full', 'started', $1, $2)`,
|
||||
[startedAt, triggeredBy]
|
||||
);
|
||||
} catch (err) {
|
||||
console.warn('[Pax8Sync] Could not create sync_history row:', err);
|
||||
}
|
||||
}
|
||||
|
||||
private async updateHistory(
|
||||
startedAt: Date,
|
||||
status: 'completed' | 'failed',
|
||||
recordsAdded: number,
|
||||
recordsDeleted: number,
|
||||
errorMessage: string | null
|
||||
): Promise<void> {
|
||||
try {
|
||||
await postgresClient.query(
|
||||
`UPDATE sync_history
|
||||
SET status = $1, completed_at = NOW(), records_added = $2, records_deleted = $3, error_message = $4
|
||||
WHERE entity_type = 'pax8' AND sync_type = 'full' AND started_at = $5`,
|
||||
[status, recordsAdded, recordsDeleted, errorMessage, startedAt]
|
||||
);
|
||||
} catch (err) {
|
||||
console.warn('[Pax8Sync] Could not update sync_history row:', err);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
let _instance: Pax8SyncService | null = null;
|
||||
|
||||
export function getPax8SyncService(): Pax8SyncService {
|
||||
if (!_instance) {
|
||||
_instance = new Pax8SyncService();
|
||||
}
|
||||
return _instance;
|
||||
}
|
||||
Loading…
Add table
Add a link
Reference in a new issue