feat(24-02): add Route53SyncService — zones and records mirror sync
- fullSync/incrementalSync + getRoute53SyncService() singleton - Paginated ListHostedZonesCommand + GetHostedZoneCommand (delegation set for D-12) - Paginated ListResourceRecordSetsCommand per live zone - Soft-delete reconciliation for zones and records (never hard-delete) - sync_history bookkeeping with entity_type='route53', literal full/incremental sync_type - Drift detection wired via buildDriftHistoryRows, writing sync_detected_drift history rows - No integration_settings gating anywhere (D-10)
This commit is contained in:
parent
c18271dda9
commit
d8c0912f4b
1 changed files with 449 additions and 0 deletions
449
lib/services/route53-sync-service.ts
Normal file
449
lib/services/route53-sync-service.ts
Normal file
|
|
@ -0,0 +1,449 @@
|
|||
/**
|
||||
* Route 53 Sync Service
|
||||
*
|
||||
* Mirrors AWS Route 53 hosted zones and resource record sets into Postgres
|
||||
* (route53_zones / route53_records), soft-deleting anything AWS no longer
|
||||
* returns, and detects drift — recording a route53_record_history row
|
||||
* (source='sync_detected_drift') for every record whose live AWS value no
|
||||
* longer matches the mirror (D-06).
|
||||
*
|
||||
* Follows lib/services/veeam-sync-service.ts's class shape: constructor
|
||||
* takes an optional client, isSyncing guard, executeSync() step loop with
|
||||
* per-step error isolation, sync_history bookkeeping.
|
||||
*/
|
||||
|
||||
import {
|
||||
Route53Client,
|
||||
ListHostedZonesCommand,
|
||||
GetHostedZoneCommand,
|
||||
ListResourceRecordSetsCommand,
|
||||
type HostedZone,
|
||||
type ResourceRecordSet,
|
||||
} from '@aws-sdk/client-route-53';
|
||||
import postgresClient from './postgres-client';
|
||||
import { getRoute53Client } from './route53-factory';
|
||||
import type { Route53SyncResult } from '@/lib/types/route53';
|
||||
import {
|
||||
buildRecordKey,
|
||||
normalizeRecordSet,
|
||||
classifyDrift,
|
||||
toHistoryPayload,
|
||||
type NormalizedRecordSet,
|
||||
} from './route53-record-key';
|
||||
|
||||
export interface Route53EntitySyncResult {
|
||||
entity: string;
|
||||
success: boolean;
|
||||
recordsUpserted: number;
|
||||
duration: number;
|
||||
error?: string;
|
||||
}
|
||||
|
||||
export interface DriftHistoryRow {
|
||||
zoneId: string;
|
||||
recordKey: string;
|
||||
recordName: string;
|
||||
recordType: string;
|
||||
changeAction: 'create' | 'update' | 'delete';
|
||||
beforeValue: unknown;
|
||||
afterValue: unknown;
|
||||
}
|
||||
|
||||
/**
|
||||
* Union the previous (mirror) and next (AWS-live) keyed recordset maps,
|
||||
* classify drift per key, and build the row shape destined for
|
||||
* route53_record_history. Pure function — no database handle — so it is
|
||||
* unit-testable without Postgres or AWS.
|
||||
*/
|
||||
export function buildDriftHistoryRows(
|
||||
prevByKey: Map<string, NormalizedRecordSet>,
|
||||
nextByKey: Map<string, NormalizedRecordSet>,
|
||||
zoneId: string
|
||||
): DriftHistoryRow[] {
|
||||
const rows: DriftHistoryRow[] = [];
|
||||
const allKeys = new Set<string>([...prevByKey.keys(), ...nextByKey.keys()]);
|
||||
|
||||
for (const key of allKeys) {
|
||||
const prev = prevByKey.get(key) ?? null;
|
||||
const next = nextByKey.get(key) ?? null;
|
||||
const action = classifyDrift(prev, next);
|
||||
if (action === null) continue;
|
||||
|
||||
const source = next ?? prev;
|
||||
if (!source) continue;
|
||||
|
||||
rows.push({
|
||||
zoneId,
|
||||
recordKey: key,
|
||||
recordName: source.name,
|
||||
recordType: source.type,
|
||||
changeAction: action,
|
||||
beforeValue: toHistoryPayload(prev),
|
||||
afterValue: toHistoryPayload(next),
|
||||
});
|
||||
}
|
||||
|
||||
return rows;
|
||||
}
|
||||
|
||||
export class Route53SyncService {
|
||||
private client: Route53Client;
|
||||
private isSyncing = false;
|
||||
|
||||
constructor(client?: Route53Client) {
|
||||
this.client = client || getRoute53Client();
|
||||
}
|
||||
|
||||
isSyncInProgress(): boolean {
|
||||
return this.isSyncing;
|
||||
}
|
||||
|
||||
/**
|
||||
* Full sync — fetches all zones/records and upserts into Postgres.
|
||||
*/
|
||||
async fullSync(triggeredBy: string = 'system'): Promise<Route53SyncResult> {
|
||||
return this.executeSync('full', triggeredBy);
|
||||
}
|
||||
|
||||
/**
|
||||
* Incremental sync — Route 53's list APIs expose no modification cursor
|
||||
* (no lastTrackedModificationDateTime analog), so an "incremental" run
|
||||
* executes the exact same full read against AWS with the same diff logic
|
||||
* as fullSync; the only difference is cadence (D-11). This is not a
|
||||
* missing optimization — Route 53 offers nothing cheaper to poll.
|
||||
*/
|
||||
async incrementalSync(triggeredBy: string = 'system'): Promise<Route53SyncResult> {
|
||||
return this.executeSync('incremental', triggeredBy);
|
||||
}
|
||||
|
||||
private async executeSync(syncType: 'full' | 'incremental', triggeredBy: string): Promise<Route53SyncResult> {
|
||||
if (this.isSyncing) {
|
||||
throw new Error('A Route 53 sync operation is already in progress');
|
||||
}
|
||||
|
||||
this.isSyncing = true;
|
||||
const syncId = `route53-${Date.now()}`;
|
||||
const startTime = new Date();
|
||||
const entityResults: Route53EntitySyncResult[] = [];
|
||||
const errors: string[] = [];
|
||||
|
||||
// Create sync_history record. The CHECK constraint on sync_type only
|
||||
// allows 'full' | 'incremental' | 'entity-specific' — write the literal
|
||||
// syncType value, not a composite string like 'route53-full'.
|
||||
let historyId: number | null = null;
|
||||
try {
|
||||
const histResult = await postgresClient.query<{ id: number }>(
|
||||
`INSERT INTO sync_history (entity_type, sync_type, status, started_at, records_added, records_updated, records_deleted, triggered_by)
|
||||
VALUES ($1, $2, $3, $4, 0, 0, 0, $5) RETURNING id`,
|
||||
['route53', syncType, 'started', startTime, triggeredBy]
|
||||
);
|
||||
historyId = histResult.rows[0].id;
|
||||
} catch (e) {
|
||||
console.warn('[ROUTE53-SYNC] Could not create sync history record:', e instanceof Error ? e.message : String(e));
|
||||
}
|
||||
|
||||
console.log(`[ROUTE53-SYNC] Starting ${syncType} sync (${syncId})`);
|
||||
|
||||
try {
|
||||
// Zones must sync before records — route53_records.zone_id has an FK
|
||||
// to route53_zones(id).
|
||||
const steps: Array<{ name: string; fn: () => Promise<number> }> = [
|
||||
{ name: 'zones', fn: () => this.syncZones() },
|
||||
{ name: 'records', fn: () => this.syncRecords() },
|
||||
];
|
||||
|
||||
for (const step of steps) {
|
||||
const stepStart = Date.now();
|
||||
try {
|
||||
const count = await step.fn();
|
||||
const duration = Date.now() - stepStart;
|
||||
entityResults.push({ entity: step.name, success: true, recordsUpserted: count, duration });
|
||||
console.log(`[ROUTE53-SYNC] ${step.name}: ${count} records in ${duration}ms`);
|
||||
} catch (error) {
|
||||
const duration = Date.now() - stepStart;
|
||||
const msg = error instanceof Error ? error.message : String(error);
|
||||
errors.push(`${step.name}: ${msg}`);
|
||||
entityResults.push({ entity: step.name, success: false, recordsUpserted: 0, duration, error: msg });
|
||||
console.error(`[ROUTE53-SYNC] ${step.name} failed:`, msg);
|
||||
}
|
||||
}
|
||||
|
||||
const completedAt = new Date();
|
||||
const duration = completedAt.getTime() - startTime.getTime();
|
||||
const status = errors.length === 0 ? 'completed' : 'failed';
|
||||
const totalRecords = entityResults.reduce((sum, r) => sum + r.recordsUpserted, 0);
|
||||
|
||||
console.log(`[ROUTE53-SYNC] Sync ${status} in ${duration}ms — ${totalRecords} total records`);
|
||||
|
||||
if (historyId) {
|
||||
try {
|
||||
await postgresClient.query(
|
||||
`UPDATE sync_history SET status = $1, completed_at = $2, records_added = $3, error_message = $4, entity_details = $5 WHERE id = $6`,
|
||||
[status, completedAt, totalRecords, errors.length > 0 ? errors.join('; ') : null, JSON.stringify(entityResults), historyId]
|
||||
);
|
||||
} catch (e) {
|
||||
console.warn('[ROUTE53-SYNC] Could not update sync history:', e instanceof Error ? e.message : String(e));
|
||||
}
|
||||
}
|
||||
|
||||
return {
|
||||
syncId,
|
||||
syncType,
|
||||
status,
|
||||
startedAt: startTime.toISOString(),
|
||||
completedAt: completedAt.toISOString(),
|
||||
duration,
|
||||
entities: { results: entityResults },
|
||||
errors,
|
||||
};
|
||||
} catch (error) {
|
||||
const completedAt = new Date();
|
||||
const msg = error instanceof Error ? error.message : String(error);
|
||||
console.error('[ROUTE53-SYNC] Sync failed catastrophically:', msg);
|
||||
|
||||
if (historyId) {
|
||||
try {
|
||||
await postgresClient.query(
|
||||
`UPDATE sync_history SET status = 'failed', completed_at = $1, error_message = $2 WHERE id = $3`,
|
||||
[completedAt, msg, historyId]
|
||||
);
|
||||
} catch {
|
||||
/* ignore */
|
||||
}
|
||||
}
|
||||
|
||||
return {
|
||||
syncId,
|
||||
syncType,
|
||||
status: 'failed',
|
||||
startedAt: startTime.toISOString(),
|
||||
completedAt: completedAt.toISOString(),
|
||||
duration: completedAt.getTime() - startTime.getTime(),
|
||||
entities: { results: entityResults },
|
||||
errors: [msg],
|
||||
};
|
||||
} finally {
|
||||
this.isSyncing = false;
|
||||
}
|
||||
}
|
||||
|
||||
// ============================================================================
|
||||
// Zone sync
|
||||
// ============================================================================
|
||||
|
||||
private async syncZones(): Promise<number> {
|
||||
const zones: HostedZone[] = [];
|
||||
let marker: string | undefined;
|
||||
|
||||
do {
|
||||
const response = await this.client.send(new ListHostedZonesCommand({ Marker: marker }));
|
||||
zones.push(...(response.HostedZones ?? []));
|
||||
marker = response.IsTruncated ? response.NextMarker : undefined;
|
||||
} while (marker);
|
||||
|
||||
const seenIds: string[] = [];
|
||||
|
||||
for (const zone of zones) {
|
||||
const id = (zone.Id ?? '').replace('/hostedzone/', '');
|
||||
if (!id) continue;
|
||||
seenIds.push(id);
|
||||
|
||||
let nameServers: string[] = [];
|
||||
try {
|
||||
const zoneDetail = await this.client.send(new GetHostedZoneCommand({ Id: id }));
|
||||
nameServers = zoneDetail.DelegationSet?.NameServers ?? [];
|
||||
} catch (error) {
|
||||
console.warn(
|
||||
`[ROUTE53-SYNC] Could not fetch delegation set for zone ${id}:`,
|
||||
error instanceof Error ? error.message : String(error)
|
||||
);
|
||||
}
|
||||
|
||||
await postgresClient.query(
|
||||
`INSERT INTO route53_zones (id, name, comment, private_zone, record_count, authoritative_name_servers, raw_payload, synced_at, updated_at, is_deleted, deleted_at)
|
||||
VALUES ($1, $2, $3, $4, $5, $6, $7, NOW(), NOW(), false, NULL)
|
||||
ON CONFLICT (id) DO UPDATE SET
|
||||
name = EXCLUDED.name, comment = EXCLUDED.comment, private_zone = EXCLUDED.private_zone,
|
||||
record_count = EXCLUDED.record_count, authoritative_name_servers = EXCLUDED.authoritative_name_servers,
|
||||
raw_payload = EXCLUDED.raw_payload, synced_at = NOW(), updated_at = NOW(),
|
||||
is_deleted = false, deleted_at = NULL`,
|
||||
[
|
||||
id,
|
||||
zone.Name ?? '',
|
||||
zone.Config?.Comment ?? null,
|
||||
zone.Config?.PrivateZone ?? false,
|
||||
zone.ResourceRecordSetCount ?? 0,
|
||||
JSON.stringify(nameServers),
|
||||
JSON.stringify(zone),
|
||||
]
|
||||
);
|
||||
}
|
||||
|
||||
// Soft-delete zones no longer returned by AWS.
|
||||
await postgresClient.query(
|
||||
`UPDATE route53_zones SET is_deleted = true, deleted_at = NOW(), updated_at = NOW()
|
||||
WHERE is_deleted = false AND id <> ALL($1)`,
|
||||
[seenIds]
|
||||
);
|
||||
|
||||
return seenIds.length;
|
||||
}
|
||||
|
||||
// ============================================================================
|
||||
// Record sync (includes drift detection)
|
||||
// ============================================================================
|
||||
|
||||
private async syncRecords(): Promise<number> {
|
||||
const liveZones = await postgresClient.query<{ id: string }>(
|
||||
'SELECT id FROM route53_zones WHERE is_deleted = false'
|
||||
);
|
||||
|
||||
let totalUpserted = 0;
|
||||
|
||||
for (const { id: zoneId } of liveZones.rows) {
|
||||
const recordsets = await this.fetchAllRecordSets(zoneId);
|
||||
const nextByKey = new Map<string, NormalizedRecordSet>();
|
||||
for (const rs of recordsets) {
|
||||
const normalized = normalizeRecordSet(rs, zoneId);
|
||||
nextByKey.set(normalized.recordKey, normalized);
|
||||
}
|
||||
|
||||
// Load the current mirror rows for this zone BEFORE upserting, so we
|
||||
// can diff against the pre-sync state (after the upsert the previous
|
||||
// state is gone).
|
||||
const mirrorRows = await postgresClient.query<{
|
||||
record_key: string;
|
||||
name: string;
|
||||
type: string;
|
||||
set_identifier: string | null;
|
||||
ttl: number | null;
|
||||
resource_records: Array<{ value: string }> | null;
|
||||
alias_target: Record<string, unknown> | null;
|
||||
}>(
|
||||
`SELECT record_key, name, type, set_identifier, ttl, resource_records, alias_target
|
||||
FROM route53_records WHERE zone_id = $1 AND is_deleted = false`,
|
||||
[zoneId]
|
||||
);
|
||||
|
||||
const prevByKey = new Map<string, NormalizedRecordSet>();
|
||||
for (const row of mirrorRows.rows) {
|
||||
prevByKey.set(row.record_key, {
|
||||
recordKey: row.record_key,
|
||||
zoneId,
|
||||
name: row.name,
|
||||
type: row.type,
|
||||
setIdentifier: row.set_identifier,
|
||||
ttl: row.ttl,
|
||||
resourceRecords: row.resource_records ?? [],
|
||||
aliasTarget: row.alias_target,
|
||||
});
|
||||
}
|
||||
|
||||
// Guard the very first sync: an empty mirror means this is the initial
|
||||
// import for the zone. Do NOT emit `create` history rows for every
|
||||
// record — that would flood the append-only ledger with thousands of
|
||||
// meaningless rows.
|
||||
if (prevByKey.size === 0) {
|
||||
console.log(`[ROUTE53-SYNC] Initial import for zone ${zoneId} — skipping drift history`);
|
||||
} else {
|
||||
const driftRows = buildDriftHistoryRows(prevByKey, nextByKey, zoneId);
|
||||
if (driftRows.length > 0) {
|
||||
await postgresClient.transaction(async (client) => {
|
||||
for (const row of driftRows) {
|
||||
await client.query(
|
||||
`INSERT INTO route53_record_history
|
||||
(zone_id, record_key, record_name, record_type, change_action, before_value, after_value, source, changed_by_user_id, changed_by_email)
|
||||
VALUES ($1, $2, $3, $4, $5, $6, $7, 'sync_detected_drift', NULL, NULL)`,
|
||||
[
|
||||
row.zoneId,
|
||||
row.recordKey,
|
||||
row.recordName,
|
||||
row.recordType,
|
||||
row.changeAction,
|
||||
JSON.stringify(row.beforeValue),
|
||||
JSON.stringify(row.afterValue),
|
||||
]
|
||||
);
|
||||
}
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
const seenKeys: string[] = [];
|
||||
for (const normalized of nextByKey.values()) {
|
||||
seenKeys.push(normalized.recordKey);
|
||||
await postgresClient.query(
|
||||
`INSERT INTO route53_records (record_key, zone_id, name, type, set_identifier, ttl, resource_records, alias_target, raw_payload, synced_at, updated_at, is_deleted, deleted_at)
|
||||
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, NOW(), NOW(), false, NULL)
|
||||
ON CONFLICT (record_key) DO UPDATE SET
|
||||
name = EXCLUDED.name, type = EXCLUDED.type, set_identifier = EXCLUDED.set_identifier,
|
||||
ttl = EXCLUDED.ttl, resource_records = EXCLUDED.resource_records, alias_target = EXCLUDED.alias_target,
|
||||
raw_payload = EXCLUDED.raw_payload, synced_at = NOW(), updated_at = NOW(),
|
||||
is_deleted = false, deleted_at = NULL`,
|
||||
[
|
||||
normalized.recordKey,
|
||||
zoneId,
|
||||
normalized.name,
|
||||
normalized.type,
|
||||
normalized.setIdentifier,
|
||||
normalized.ttl,
|
||||
JSON.stringify(normalized.resourceRecords),
|
||||
JSON.stringify(normalized.aliasTarget),
|
||||
JSON.stringify(normalized),
|
||||
]
|
||||
);
|
||||
totalUpserted++;
|
||||
}
|
||||
|
||||
// Soft-delete this zone's records no longer present in AWS. Never
|
||||
// hard-delete — the history ledger references record_key.
|
||||
await postgresClient.query(
|
||||
`UPDATE route53_records SET is_deleted = true, deleted_at = NOW(), updated_at = NOW()
|
||||
WHERE zone_id = $1 AND is_deleted = false AND record_key <> ALL($2)`,
|
||||
[zoneId, seenKeys]
|
||||
);
|
||||
}
|
||||
|
||||
return totalUpserted;
|
||||
}
|
||||
|
||||
private async fetchAllRecordSets(zoneId: string): Promise<ResourceRecordSet[]> {
|
||||
const recordsets: ResourceRecordSet[] = [];
|
||||
let startRecordName: string | undefined;
|
||||
let startRecordType: ResourceRecordSet['Type'] | undefined;
|
||||
let startRecordIdentifier: string | undefined;
|
||||
|
||||
do {
|
||||
const response = await this.client.send(
|
||||
new ListResourceRecordSetsCommand({
|
||||
HostedZoneId: zoneId,
|
||||
StartRecordName: startRecordName,
|
||||
StartRecordType: startRecordType,
|
||||
StartRecordIdentifier: startRecordIdentifier,
|
||||
})
|
||||
);
|
||||
recordsets.push(...(response.ResourceRecordSets ?? []));
|
||||
|
||||
if (response.IsTruncated) {
|
||||
startRecordName = response.NextRecordName;
|
||||
startRecordType = response.NextRecordType;
|
||||
startRecordIdentifier = response.NextRecordIdentifier;
|
||||
} else {
|
||||
startRecordName = undefined;
|
||||
startRecordType = undefined;
|
||||
startRecordIdentifier = undefined;
|
||||
}
|
||||
} while (startRecordName);
|
||||
|
||||
return recordsets;
|
||||
}
|
||||
}
|
||||
|
||||
let _instance: Route53SyncService | null = null;
|
||||
|
||||
export function getRoute53SyncService(): Route53SyncService {
|
||||
if (!_instance) {
|
||||
_instance = new Route53SyncService();
|
||||
}
|
||||
return _instance;
|
||||
}
|
||||
Loading…
Add table
Add a link
Reference in a new issue