331 lines
10 KiB
TypeScript
331 lines
10 KiB
TypeScript
/**
|
|
* Mimecast Sync Service
|
|
* Full sync (120 days back), incremental (since last sync), body fetch, 120-day purge
|
|
*/
|
|
|
|
import { getMimecastClient, MimecastMessage, MimecastThreatEvent } from './mimecast-client';
|
|
import { postgresClient as pg } from './postgres-client';
|
|
|
|
export interface MimecastSyncResult {
|
|
messagesUpserted: number;
|
|
threatsUpserted: number;
|
|
bodiesFetched: number;
|
|
purgedMessages: number;
|
|
errors: string[];
|
|
durationMs: number;
|
|
}
|
|
|
|
const RETENTION_DAYS = 120;
|
|
const BODY_FETCH_LIMIT = 500;
|
|
|
|
// ── Upsert helpers ────────────────────────────────────────────────────────────
|
|
|
|
async function upsertMessages(messages: MimecastMessage[]): Promise<number> {
|
|
if (!messages.length) return 0;
|
|
let count = 0;
|
|
|
|
for (const m of messages) {
|
|
if (!m.id) continue;
|
|
const senderDomain = m.senderAddress?.includes('@')
|
|
? m.senderAddress.split('@')[1]?.toLowerCase()
|
|
: null;
|
|
|
|
await pg.query(`
|
|
INSERT INTO mimecast_messages (
|
|
id, sender_address, sender_domain, recipient_address, subject,
|
|
direction, status, action, spam_score, size_bytes, attachment_count,
|
|
sent_datetime, received_datetime, delivery_datetime,
|
|
route, reject_reason, held_reason, source_ip, raw, synced_at
|
|
) VALUES (
|
|
$1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16,$17,$18,$19,NOW()
|
|
)
|
|
ON CONFLICT (id) DO UPDATE SET
|
|
status = EXCLUDED.status,
|
|
action = EXCLUDED.action,
|
|
delivery_datetime= EXCLUDED.delivery_datetime,
|
|
reject_reason = EXCLUDED.reject_reason,
|
|
held_reason = EXCLUDED.held_reason,
|
|
raw = EXCLUDED.raw,
|
|
synced_at = NOW()
|
|
`, [
|
|
m.id,
|
|
m.senderAddress || null,
|
|
senderDomain,
|
|
m.recipientAddress || null,
|
|
m.subject || null,
|
|
m.direction || null,
|
|
m.status || null,
|
|
m.action || null,
|
|
m.spamScore ?? null,
|
|
m.sizeBytes ?? null,
|
|
typeof m.attachmentCount === 'number' ? m.attachmentCount : null,
|
|
m.sentDateTime || null,
|
|
m.receivedDateTime || null,
|
|
null, // delivery_datetime not in message-finder search
|
|
m.route || null,
|
|
m.rejectReason || null,
|
|
m.heldReason || null,
|
|
m.sourceIp || null,
|
|
m._raw ? JSON.stringify(m._raw) : null,
|
|
]);
|
|
count++;
|
|
}
|
|
|
|
return count;
|
|
}
|
|
|
|
async function upsertThreats(events: MimecastThreatEvent[]): Promise<number> {
|
|
if (!events.length) return 0;
|
|
let count = 0;
|
|
|
|
for (const e of events) {
|
|
if (!e.id) continue;
|
|
await pg.query(`
|
|
INSERT INTO mimecast_threat_events (
|
|
id, message_id, event_type, threat_level, url, file_name,
|
|
verdict, actor_email, event_datetime, details, synced_at
|
|
) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,NOW())
|
|
ON CONFLICT (id) DO UPDATE SET
|
|
threat_level = EXCLUDED.threat_level,
|
|
verdict = EXCLUDED.verdict,
|
|
details = EXCLUDED.details,
|
|
synced_at = NOW()
|
|
`, [
|
|
e.id,
|
|
e.messageId || null,
|
|
e.eventType || null,
|
|
e.threatLevel || null,
|
|
e.url || null,
|
|
e.fileName || null,
|
|
e.verdict || null,
|
|
e.actorEmail || null,
|
|
e.eventDateTime || null,
|
|
e._raw ? JSON.stringify(e._raw) : null,
|
|
]);
|
|
count++;
|
|
}
|
|
|
|
return count;
|
|
}
|
|
|
|
async function fetchAndStoreBodies(errors: string[]): Promise<number> {
|
|
const client = getMimecastClient();
|
|
|
|
// Find delivered inbound messages missing body, up to limit
|
|
const rows = await pg.query(`
|
|
SELECT m.id FROM mimecast_messages m
|
|
LEFT JOIN mimecast_message_bodies b ON b.message_id = m.id
|
|
WHERE b.message_id IS NULL
|
|
AND m.status NOT IN ('rejected','spam','bounced')
|
|
AND m.direction = 'inbound'
|
|
ORDER BY m.sent_datetime DESC
|
|
LIMIT $1
|
|
`, [BODY_FETCH_LIMIT]);
|
|
|
|
let fetched = 0;
|
|
for (const row of rows.rows) {
|
|
try {
|
|
const info = await client.getMessageInfo(row.id);
|
|
if (!info) continue;
|
|
|
|
await pg.query(`
|
|
INSERT INTO mimecast_message_bodies (message_id, body_text, body_html, headers, fetched_at)
|
|
VALUES ($1, $2, $3, $4, NOW())
|
|
ON CONFLICT (message_id) DO UPDATE SET
|
|
body_text = EXCLUDED.body_text,
|
|
body_html = EXCLUDED.body_html,
|
|
headers = EXCLUDED.headers,
|
|
fetched_at = NOW()
|
|
`, [
|
|
info.messageId,
|
|
info.bodyText || null,
|
|
info.bodyHtml || null,
|
|
info.headers ? JSON.stringify(info.headers) : null,
|
|
]);
|
|
fetched++;
|
|
} catch (err: any) {
|
|
errors.push(`Body fetch ${row.id}: ${err.message}`);
|
|
}
|
|
}
|
|
|
|
return fetched;
|
|
}
|
|
|
|
async function purgeOldRecords(): Promise<number> {
|
|
const cutoff = new Date();
|
|
cutoff.setDate(cutoff.getDate() - RETENTION_DAYS);
|
|
|
|
const result = await pg.query(
|
|
`DELETE FROM mimecast_messages WHERE sent_datetime < $1`,
|
|
[cutoff.toISOString()]
|
|
);
|
|
return result.rowCount ?? 0;
|
|
}
|
|
|
|
async function getSyncState(syncType: string): Promise<{ lastSyncedAt: Date | null; cursor: string | null }> {
|
|
const result = await pg.query(
|
|
`SELECT last_synced_at, cursor FROM mimecast_sync_state WHERE sync_type = $1`,
|
|
[syncType]
|
|
);
|
|
const row = result.rows[0];
|
|
return {
|
|
lastSyncedAt: row?.last_synced_at ? new Date(row.last_synced_at) : null,
|
|
cursor: row?.cursor ?? null,
|
|
};
|
|
}
|
|
|
|
async function saveSyncState(syncType: string, lastSyncedAt: Date, cursor: string | null): Promise<void> {
|
|
await pg.query(`
|
|
INSERT INTO mimecast_sync_state (sync_type, last_synced_at, cursor, updated_at)
|
|
VALUES ($1, $2, $3, NOW())
|
|
ON CONFLICT (sync_type) DO UPDATE SET
|
|
last_synced_at = EXCLUDED.last_synced_at,
|
|
cursor = EXCLUDED.cursor,
|
|
updated_at = NOW()
|
|
`, [syncType, lastSyncedAt.toISOString(), cursor]);
|
|
}
|
|
|
|
// ── Main sync functions ───────────────────────────────────────────────────────
|
|
|
|
export async function syncMimecastMessages(
|
|
fromDate: Date,
|
|
toDate: Date,
|
|
errors: string[]
|
|
): Promise<number> {
|
|
const client = getMimecastClient();
|
|
let total = 0;
|
|
let cursor: string | null = null;
|
|
|
|
const from = fromDate.toISOString();
|
|
const to = toDate.toISOString();
|
|
|
|
do {
|
|
try {
|
|
const result = await client.getMessageLogs({ from, to, cursor: cursor ?? undefined, pageSize: 500 });
|
|
if (result.items.length > 0) {
|
|
total += await upsertMessages(result.items);
|
|
}
|
|
cursor = result.nextCursor;
|
|
} catch (err: any) {
|
|
errors.push(`Message sync page: ${err.message}`);
|
|
break;
|
|
}
|
|
} while (cursor);
|
|
|
|
await saveSyncState('messages', toDate, null);
|
|
return total;
|
|
}
|
|
|
|
export async function syncMimecastThreats(errors: string[]): Promise<number> {
|
|
const client = getMimecastClient();
|
|
let total = 0;
|
|
let cursor: string | null = null;
|
|
|
|
do {
|
|
try {
|
|
const result = await client.getThreatEvents({ cursor: cursor ?? undefined, pageSize: 500 });
|
|
if (result.items.length > 0) {
|
|
total += await upsertThreats(result.items);
|
|
}
|
|
cursor = result.nextCursor;
|
|
} catch (err: any) {
|
|
errors.push(`Threat sync: ${err.message}`);
|
|
break;
|
|
}
|
|
} while (cursor);
|
|
|
|
await saveSyncState('threats', new Date(), null);
|
|
return total;
|
|
}
|
|
|
|
// ── Public API ────────────────────────────────────────────────────────────────
|
|
|
|
export async function runMimecastFullSync(): Promise<MimecastSyncResult> {
|
|
const start = Date.now();
|
|
const errors: string[] = [];
|
|
|
|
const toDate = new Date();
|
|
const fromDate = new Date();
|
|
fromDate.setDate(fromDate.getDate() - RETENTION_DAYS);
|
|
|
|
const messagesUpserted = await syncMimecastMessages(fromDate, toDate, errors);
|
|
const threatsUpserted = await syncMimecastThreats(errors);
|
|
const bodiesFetched = await fetchAndStoreBodies(errors);
|
|
const purgedMessages = await purgeOldRecords();
|
|
|
|
return {
|
|
messagesUpserted,
|
|
threatsUpserted,
|
|
bodiesFetched,
|
|
purgedMessages,
|
|
errors,
|
|
durationMs: Date.now() - start,
|
|
};
|
|
}
|
|
|
|
export async function runMimecastIncrementalSync(): Promise<MimecastSyncResult> {
|
|
const start = Date.now();
|
|
const errors: string[] = [];
|
|
|
|
const { lastSyncedAt } = await getSyncState('messages');
|
|
|
|
const toDate = new Date();
|
|
const fromDate = lastSyncedAt ?? (() => {
|
|
const d = new Date();
|
|
d.setDate(d.getDate() - 1);
|
|
return d;
|
|
})();
|
|
|
|
const messagesUpserted = await syncMimecastMessages(fromDate, toDate, errors);
|
|
const threatsUpserted = await syncMimecastThreats(errors);
|
|
const bodiesFetched = await fetchAndStoreBodies(errors);
|
|
const purgedMessages = await purgeOldRecords();
|
|
|
|
return {
|
|
messagesUpserted,
|
|
threatsUpserted,
|
|
bodiesFetched,
|
|
purgedMessages,
|
|
errors,
|
|
durationMs: Date.now() - start,
|
|
};
|
|
}
|
|
|
|
export async function getMimecastStats(): Promise<{
|
|
messages: number;
|
|
inbound: number;
|
|
outbound: number;
|
|
threats: number;
|
|
bodies: number;
|
|
lastSync: string | null;
|
|
oldestMessage: string | null;
|
|
}> {
|
|
const [counts, syncState, oldest] = await Promise.all([
|
|
pg.query(`
|
|
SELECT
|
|
COUNT(*) AS total,
|
|
COUNT(*) FILTER (WHERE direction = 'inbound') AS inbound,
|
|
COUNT(*) FILTER (WHERE direction = 'outbound') AS outbound
|
|
FROM mimecast_messages
|
|
`),
|
|
pg.query(`SELECT last_synced_at FROM mimecast_sync_state WHERE sync_type = 'messages'`),
|
|
pg.query(`SELECT MIN(sent_datetime) AS oldest FROM mimecast_messages`),
|
|
]);
|
|
|
|
const [threatCount, bodyCount] = await Promise.all([
|
|
pg.query(`SELECT COUNT(*) AS total FROM mimecast_threat_events`),
|
|
pg.query(`SELECT COUNT(*) AS total FROM mimecast_message_bodies`),
|
|
]);
|
|
|
|
const c = counts.rows[0] ?? {};
|
|
|
|
return {
|
|
messages: Number(c.total ?? 0),
|
|
inbound: Number(c.inbound ?? 0),
|
|
outbound: Number(c.outbound ?? 0),
|
|
threats: Number(threatCount.rows[0]?.total ?? 0),
|
|
bodies: Number(bodyCount.rows[0]?.total ?? 0),
|
|
lastSync: syncState.rows[0]?.last_synced_at ?? null,
|
|
oldestMessage: oldest.rows[0]?.oldest ?? null,
|
|
};
|
|
}
|