feat(11-02): add Pax8SyncService (companies/subscriptions/referenced catalog)
- fullSync() orchestrates companies -> subscriptions -> referenced-only products, each with independent try/catch returning Pax8EntitySyncResult - Referenced product catalog resolved in a single pass: listAllProducts() fetched once, filtered in-memory to subscription-referenced ids (D-01/D-02) - Each entity's UUID-array tombstone soft-deletes rows PAX8 no longer returns (is_deleted=true, deleted_at set), skipped when zero ids seen - sync_history row (entity_type='pax8', sync_type='full') tracks started/completed/failed with base columns only - No PAX8 writes: only Pax8Client's read methods are called
This commit is contained in:
parent
6c64aaf7bb
commit
cf3ae61dcb
1 changed files with 368 additions and 0 deletions
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