fix: record venue credit time on deposit archive rows

src/handlers/execute-action/deposit.ts

@@ -3,6 +3,7 @@ import ccxt from "@usherlabs/ccxt";
 import {
 	archiveTransferEventInBackground,
 	normalizeCcxtTransactionForArchive,
+	normalizeTimestamp,
 } from "../../helpers/broker-execution-archive";
 import {
 	depositField,
@@ -160,8 +161,7 @@ export async function handleDeposit(ctx: ExecuteActionContext): Promise<void> {
 					network: depositNetwork?.exchangeNetworkId,
 					externalId: depositTxid,
 					txid: depositTxid,
-					exchangeTimestamp:
-						typeof creditedAt === "string" ? creditedAt : undefined,
+					exchangeTimestamp: normalizeTimestamp(creditedAt),
 					payload: deposit,
 				},
 			});

src/helpers/broker-execution-archive/index.ts

@@ -27,6 +27,7 @@ export {
 	normalizeCcxtBalanceForArchive,
 	normalizeCcxtTradeForArchive,
 	normalizeCcxtTransactionForArchive,
+	normalizeTimestamp,
 	type TransferArchiveFields,
 } from "./rows";
 export {

src/helpers/broker-execution-archive/rows.ts

@@ -58,7 +58,12 @@ function firstNumber(...values: unknown[]): number | undefined {
 	return undefined;
 }
 
-function normalizeTimestamp(value: unknown): string | undefined {
+// Venues report a moment as either an ISO-8601 string or an epoch-ms number
+// (Binance sends `insertTime` as an integer). The archive's DateTime64 columns
+// are fed ISO-8601 UTC strings — the forwarder configures ClickHouse for exactly
+// that form — so an epoch-ms integer must be converted, never passed through: it
+// would be read as *seconds* and land the row tens of thousands of years ahead.
+export function normalizeTimestamp(value: unknown): string | undefined {
 	if (typeof value === "string" && value.trim()) {
 		return value.trim();
 	}

src/helpers/deposit-archive-poller.ts

@@ -4,6 +4,7 @@ import {
 	buildCommonArchiveTags,
 	buildTransferEventArchiveRow,
 	normalizeCcxtTransactionForArchive,
+	normalizeTimestamp,
 	rethrowArchiveDurabilityError,
 } from "./broker-execution-archive";
 import { depositField, normalizeDepositStatus } from "./deposit";
@@ -311,8 +312,7 @@ export class DepositArchivePoller {
 						network: network === undefined ? undefined : String(network),
 						externalId: depositTxid,
 						txid: depositTxid,
-						exchangeTimestamp:
-							typeof creditedAt === "string" ? creditedAt : undefined,
+						exchangeTimestamp: normalizeTimestamp(creditedAt),
 						payload: record,
 					},
 				}),

test/broker-execution-archive.test.ts

@@ -14,6 +14,7 @@ import type { LogRecord } from "@opentelemetry/api-logs";
 import type { Exchange } from "@usherlabs/ccxt";
 import { MAX_ARCHIVE_BODY_BYTES } from "../services/archive-forwarder/limits";
 import type { ExecuteActionContext } from "../src/handlers/execute-action/context";
+import { handleDeposit } from "../src/handlers/execute-action/deposit";
 import { handleInternalTransfer } from "../src/handlers/execute-action/internal-transfer";
 import { handleOrders } from "../src/handlers/execute-action/orders";
 import { handleTreasuryCall } from "../src/handlers/execute-action/treasury-call";
@@ -1719,3 +1720,79 @@ describe("broker execution archiver env", () => {
 		}
 	});
 });
+
+describe("deposit observation archive", () => {
+	test("records the venue credit time when the venue reports it as an epoch integer", async () => {
+		const forwarder = await startForwarderServer();
+		const archiver = BrokerExecutionArchiver.create({
+			forwarderUrl: forwarder.url,
+			deadLetterPath: createDeadLetterPath(),
+			deploymentId: "test-deploy",
+			batchSize: 100,
+			flushIntervalMs: 60_000,
+		});
+		// Binance reports insertTime as an integer, which ccxt surfaces as
+		// `timestamp`; `datetime` is absent from this shape on purpose.
+		const insertTime = 1_784_000_000_123;
+		const broker = {
+			has: { fetchDeposits: true },
+			fetchDeposits: async () => [
+				{
+					txid: "0xdeposited",
+					currency: "USDC",
+					amount: 10,
+					address: "0xrecipient",
+					status: "ok",
+					timestamp: insertTime,
+					info: { status: "1", insertTime },
+				},
+			],
+		} as unknown as Exchange;
+
+		const context = {
+			call: {
+				request: {
+					payload: {
+						recipientAddress: "0xrecipient",
+						amount: "10",
+						transactionHash: "0xdeposited",
+					},
+				},
+			},
+			wrappedCallback: () => {},
+			policy: { deposit: {} },
+			brokers: {},
+			metadata: {},
+			normalizedCex: "binance",
+			cex: "binance",
+			symbol: "USDC",
+			selectedBrokerAccount: { exchange: broker, label: "primary" },
+			broker,
+			verity: { proof: "" },
+			applyVerityToBroker: () => {},
+			useVerity: false,
+			verityProverUrl: "",
+			brokerArchiver: archiver,
+		} as unknown as ExecuteActionContext;
+
+		try {
+			await handleDeposit(context);
+			await Promise.resolve();
+			await archiver.flush();
+
+			const rows = forwarder.requests.flatMap(
+				(request) => request.body.rows ?? [],
+			) as Array<{ row: Record<string, unknown> }>;
+			expect(rows).toHaveLength(1);
+			expect(rows[0]?.row).toMatchObject({
+				event_kind: "deposit",
+				lifecycle_action: "observe_deposit",
+				external_id: "0xdeposited",
+				exchange_timestamp: new Date(insertTime).toISOString(),
+			});
+		} finally {
+			await archiver.close();
+			await forwarder.close();
+		}
+	});
+});

test/deposit-archive-poller.test.ts

@@ -119,6 +119,38 @@ describe("DepositArchivePoller.pollAllOnce", () => {
 		});
 	});
 
+	test("records the venue credit time when the venue reports it as an epoch integer", async () => {
+		// Binance reports insertTime as an integer, which ccxt surfaces as
+		// `timestamp`; `datetime` is absent from this shape on purpose.
+		const insertTime = 1_784_000_000_123;
+		const exchange = {
+			has: { fetchDeposits: true },
+			fetchDeposits: async () => [
+				{
+					txid: "0xinteger-credit",
+					currency: "USDC",
+					amount: "12",
+					status: "ok",
+					timestamp: insertTime,
+					info: { status: "1", insertTime },
+				},
+			],
+		};
+		const sink: BrokerArchiveRow[] = [];
+		const poller = new DepositArchivePoller({
+			brokers: poolWith(exchange),
+			archiver: fakeArchiver(sink),
+		});
+
+		await poller.pollAllOnce();
+
+		expect(sink).toHaveLength(1);
+		expect(sink[0]?.row).toMatchObject({
+			external_id: "0xinteger-credit",
+			exchange_timestamp: new Date(insertTime).toISOString(),
+		});
+	});
+
 	test("advances the account cursor and does not re-archive within a session", async () => {
 		const depositTimestamp = Date.now() + 10_000;
 		const deposit = {