From f2615fbe514c48557593abe58f00ae175a278fc8 Mon Sep 17 00:00:00 2001 From: adewunmiodunlamipriscilla-glitch Date: Sun, 27 Sep 2026 17:45:16 +0100 Subject: [PATCH] feat: enhance event handling with deduplication and new event structures --- .../migrations/0001_analytics_schema.sql | 122 +++++++++++++++++- .../src/lib/__tests__/contract-events.test.ts | 30 ++++- frontend/src/lib/contract.ts | 46 ++++++- .../results.json | 1 + scripts/indexer.js | 30 ++++- 5 files changed, 215 insertions(+), 14 deletions(-) create mode 100644 node_modules/.vite/vitest/da39a3ee5e6b4b0d3255bfef95601890afd80709/results.json diff --git a/analytics-api/migrations/0001_analytics_schema.sql b/analytics-api/migrations/0001_analytics_schema.sql index 98920bba..e1600f49 100644 --- a/analytics-api/migrations/0001_analytics_schema.sql +++ b/analytics-api/migrations/0001_analytics_schema.sql @@ -21,16 +21,13 @@ CREATE TABLE IF NOT EXISTS proposals ( PRIMARY KEY (contract_id, proposal_id) ); -CREATE INDEX IF NOT EXISTS proposals_created_at_idx ON proposals (created_at DESC); -CREATE INDEX IF NOT EXISTS proposals_status_category_idx ON proposals (status, category); -CREATE INDEX IF NOT EXISTS proposals_proposer_idx ON proposals (proposer); - CREATE TABLE IF NOT EXISTS events ( contract_id TEXT NOT NULL, ledger BIGINT NOT NULL CHECK (ledger >= 0), tx_hash TEXT NOT NULL, event_index INTEGER NOT NULL CHECK (event_index >= 0), proposal_id BIGINT, + schedule_id BIGINT, topic TEXT NOT NULL, actor TEXT NOT NULL DEFAULT '', occurred_at TIMESTAMPTZ NOT NULL, @@ -38,5 +35,118 @@ CREATE TABLE IF NOT EXISTS events ( PRIMARY KEY (contract_id, ledger, tx_hash, event_index) ); -CREATE INDEX IF NOT EXISTS events_proposal_timeline_idx - ON events (contract_id, proposal_id, ledger, event_index); \ No newline at end of file +CREATE TABLE IF NOT EXISTS proposal_approvals ( + contract_id TEXT NOT NULL, + proposal_id BIGINT NOT NULL CHECK (proposal_id >= 0), + owner TEXT NOT NULL, + weight INTEGER NOT NULL CHECK (weight >= 0), + approval_count INTEGER NOT NULL CHECK (approval_count >= 0), + ledger BIGINT NOT NULL CHECK (ledger >= 0), + tx_hash TEXT NOT NULL, + event_index INTEGER NOT NULL CHECK (event_index >= 0), + occurred_at TIMESTAMPTZ NOT NULL, + PRIMARY KEY (contract_id, proposal_id, owner, ledger, tx_hash, event_index) +); + +CREATE TABLE IF NOT EXISTS proposal_executions ( + contract_id TEXT NOT NULL, + proposal_id BIGINT NOT NULL CHECK (proposal_id >= 0), + executor TEXT NOT NULL, + ledger BIGINT NOT NULL CHECK (ledger >= 0), + tx_hash TEXT NOT NULL, + event_index INTEGER NOT NULL CHECK (event_index >= 0), + occurred_at TIMESTAMPTZ NOT NULL, + data JSONB NOT NULL DEFAULT '{}'::jsonb, + PRIMARY KEY (contract_id, proposal_id, ledger, tx_hash, event_index) +); + +CREATE TABLE IF NOT EXISTS transfers ( + contract_id TEXT NOT NULL, + proposal_id BIGINT, + schedule_id BIGINT, + recipient TEXT NOT NULL DEFAULT '', + token TEXT NOT NULL DEFAULT 'XLM', + amount TEXT NOT NULL DEFAULT '0', + ledger BIGINT NOT NULL CHECK (ledger >= 0), + tx_hash TEXT NOT NULL, + event_index INTEGER NOT NULL CHECK (event_index >= 0), + occurred_at TIMESTAMPTZ NOT NULL, + data JSONB NOT NULL DEFAULT '{}'::jsonb, + PRIMARY KEY (contract_id, ledger, tx_hash, event_index) +); + +CREATE TABLE IF NOT EXISTS owner_weight_changes ( + contract_id TEXT NOT NULL, + owner TEXT NOT NULL, + old_weight INTEGER NOT NULL CHECK (old_weight >= 0), + new_weight INTEGER NOT NULL CHECK (new_weight >= 0), + new_total_weight INTEGER NOT NULL CHECK (new_total_weight >= 0), + ledger BIGINT NOT NULL CHECK (ledger >= 0), + tx_hash TEXT NOT NULL, + event_index INTEGER NOT NULL CHECK (event_index >= 0), + occurred_at TIMESTAMPTZ NOT NULL, + PRIMARY KEY (contract_id, owner, ledger, tx_hash, event_index) +); + +CREATE TABLE IF NOT EXISTS recurring_disbursements ( + contract_id TEXT NOT NULL, + schedule_id BIGINT NOT NULL CHECK (schedule_id >= 0), + recipient TEXT NOT NULL DEFAULT '', + token TEXT NOT NULL DEFAULT 'XLM', + amount TEXT NOT NULL DEFAULT '0', + total_disbursed TEXT NOT NULL DEFAULT '0', + periods_disbursed INTEGER NOT NULL DEFAULT 0 CHECK (periods_disbursed >= 0), + ledger BIGINT NOT NULL CHECK (ledger >= 0), + tx_hash TEXT NOT NULL, + event_index INTEGER NOT NULL CHECK (event_index >= 0), + occurred_at TIMESTAMPTZ NOT NULL, + PRIMARY KEY (contract_id, schedule_id, ledger, tx_hash, event_index) +); + +CREATE TABLE IF NOT EXISTS delegations ( + contract_id TEXT NOT NULL, + delegator TEXT NOT NULL, + delegate TEXT NOT NULL, + weight INTEGER NOT NULL CHECK (weight >= 0), + expiry TIMESTAMPTZ, + is_active BOOLEAN NOT NULL DEFAULT true, + ledger BIGINT NOT NULL CHECK (ledger >= 0), + tx_hash TEXT NOT NULL, + event_index INTEGER NOT NULL CHECK (event_index >= 0), + occurred_at TIMESTAMPTZ NOT NULL, + PRIMARY KEY (contract_id, delegator, ledger, tx_hash, event_index) +); + +CREATE TABLE IF NOT EXISTS role_changes ( + contract_id TEXT NOT NULL, + target TEXT NOT NULL, + role_name TEXT NOT NULL, + action TEXT NOT NULL CHECK (action IN ('granted', 'revoked')), + ledger BIGINT NOT NULL CHECK (ledger >= 0), + tx_hash TEXT NOT NULL, + event_index INTEGER NOT NULL CHECK (event_index >= 0), + occurred_at TIMESTAMPTZ NOT NULL, + data JSONB NOT NULL DEFAULT '{}'::jsonb, + PRIMARY KEY (contract_id, target, role_name, ledger, tx_hash, event_index) +); + +CREATE TABLE IF NOT EXISTS checkpoints ( + contract_id TEXT PRIMARY KEY, + last_ledger BIGINT NOT NULL DEFAULT 0 CHECK (last_ledger >= 0), + updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW() +); + +CREATE INDEX IF NOT EXISTS proposals_created_at_idx ON proposals (created_at DESC); +CREATE INDEX IF NOT EXISTS proposals_status_category_idx ON proposals (status, category); +CREATE INDEX IF NOT EXISTS proposals_proposer_idx ON proposals (proposer); +CREATE INDEX IF NOT EXISTS events_contract_ledger_event_idx ON events (contract_id, ledger, event_index); +CREATE INDEX IF NOT EXISTS events_proposal_timeline_idx ON events (contract_id, proposal_id, ledger, event_index); +CREATE INDEX IF NOT EXISTS events_schedule_timeline_idx ON events (contract_id, schedule_id, ledger, event_index); +CREATE INDEX IF NOT EXISTS approvals_proposal_idx ON proposal_approvals (contract_id, proposal_id, ledger, event_index); +CREATE INDEX IF NOT EXISTS executions_proposal_idx ON proposal_executions (contract_id, proposal_id, ledger, event_index); +CREATE INDEX IF NOT EXISTS transfers_proposal_idx ON transfers (contract_id, proposal_id, ledger, event_index); +CREATE INDEX IF NOT EXISTS transfers_schedule_idx ON transfers (contract_id, schedule_id, ledger, event_index); +CREATE INDEX IF NOT EXISTS owner_weight_changes_owner_idx ON owner_weight_changes (contract_id, owner, ledger, event_index); +CREATE INDEX IF NOT EXISTS recurring_disbursements_schedule_idx ON recurring_disbursements (contract_id, schedule_id, ledger, event_index); +CREATE INDEX IF NOT EXISTS delegations_delegate_idx ON delegations (contract_id, delegate, ledger, event_index); +CREATE INDEX IF NOT EXISTS role_changes_target_idx ON role_changes (contract_id, target, ledger, event_index); \ No newline at end of file diff --git a/frontend/src/lib/__tests__/contract-events.test.ts b/frontend/src/lib/__tests__/contract-events.test.ts index 99f0d9c2..24e2af01 100644 --- a/frontend/src/lib/__tests__/contract-events.test.ts +++ b/frontend/src/lib/__tests__/contract-events.test.ts @@ -1,7 +1,18 @@ import { describe, test, expect, vi, beforeEach } from "vitest"; -import { getLatestLedger, getContractEvents, mapProposal } from "../contract"; +import { + getLatestLedger, + getContractEvents, + getProposalEvents, + mapProposal, + dedupeContractEvents, +} from "../contract"; import { rpc } from "@stellar/stellar-sdk"; +const { mockGetLatestLedger, mockGetEvents } = vi.hoisted(() => ({ + mockGetLatestLedger: vi.fn(), + mockGetEvents: vi.fn(), +})); + // Mock the rpc.Server instance directly through vi vi.mock("@stellar/stellar-sdk", async (importOriginal) => { const actual: any = await importOriginal(); @@ -213,7 +224,20 @@ describe("Contract Events API", () => { expect(proposal.to).toBe("GOWNER...1111"); expect(proposal.amount).toBe("25"); expect(proposal.token).toBe("Owner weight"); -}); + }); + + test("deduplicates overlapping event polls by ledger, tx hash, and event index", () => { + const events = [ + { ledger: 200, txHash: "tx-b", eventIndex: 2, value: "later" }, + { ledger: 100, txHash: "tx-a", eventIndex: 1, value: "first" }, + { ledger: 100, txHash: "tx-a", eventIndex: 1, value: "duplicate" }, + { ledger: 150, txHash: "tx-c", eventIndex: 0, value: "middle" }, + ]; -// TODO: Add a test proving that replaying a ledger range never double-counts events (idempotency) + const deduped = dedupeContractEvents(events); + + expect(deduped).toHaveLength(3); + expect(deduped.map((event) => event.value)).toEqual(["first", "middle", "later"]); + }); +}); diff --git a/frontend/src/lib/contract.ts b/frontend/src/lib/contract.ts index a7140095..2665cd95 100644 --- a/frontend/src/lib/contract.ts +++ b/frontend/src/lib/contract.ts @@ -823,6 +823,47 @@ function formatEventTimestamp( return "Just now"; } +export function canonicalEventKey(event: { + ledger?: number | string; + txHash?: string; + tx_hash?: string; + eventIndex?: number | string | bigint; + event_index?: number | string | bigint; +}): string { + const ledger = Number(event?.ledger ?? 0); + const txHash = String(event?.txHash ?? event?.tx_hash ?? ""); + const eventIndex = Number(event?.eventIndex ?? event?.event_index ?? 0); + return `${ledger}:${txHash}:${eventIndex}`; +} + +export function dedupeContractEvents(events: T[]): T[] { + const seen = new Set(); + return [...events] + .filter((event) => { + const key = canonicalEventKey(event); + if (seen.has(key)) return false; + seen.add(key); + return true; + }) + .sort((a, b) => { + const ledgerDelta = Number(a.ledger ?? 0) - Number(b.ledger ?? 0); + if (ledgerDelta !== 0) return ledgerDelta; + const eventDelta = + Number(a.eventIndex ?? a.event_index ?? 0) - + Number(b.eventIndex ?? b.event_index ?? 0); + if (eventDelta !== 0) return eventDelta; + return String(a.txHash ?? a.tx_hash ?? "").localeCompare( + String(b.txHash ?? b.tx_hash ?? ""), + ); + }); +} + function resolveEventType( first: string, second: string, @@ -945,9 +986,10 @@ export async function getProposalEvents( }); const events: ProposalEvent[] = []; + const rawEvents = dedupeContractEvents(res.events ?? []); - if (res.events && Array.isArray(res.events)) { - for (const rawEv of res.events) { + if (rawEvents.length > 0) { + for (const rawEv of rawEvents) { try { const rawTopic = Array.isArray(rawEv.topic) ? rawEv.topic diff --git a/node_modules/.vite/vitest/da39a3ee5e6b4b0d3255bfef95601890afd80709/results.json b/node_modules/.vite/vitest/da39a3ee5e6b4b0d3255bfef95601890afd80709/results.json new file mode 100644 index 00000000..5218e194 --- /dev/null +++ b/node_modules/.vite/vitest/da39a3ee5e6b4b0d3255bfef95601890afd80709/results.json @@ -0,0 +1 @@ +{"version":"5.0.2","results":[[":frontend/src/lib/__tests__/contract-events.test.ts",{"duration":0,"failed":true}]]} \ No newline at end of file diff --git a/scripts/indexer.js b/scripts/indexer.js index 2c910ca4..b00abc88 100755 --- a/scripts/indexer.js +++ b/scripts/indexer.js @@ -226,6 +226,25 @@ function parseVal(scVal) { } } +function canonicalEventIdentity(rawEvent, fallbackIndex = 0) { + const ledger = Number(rawEvent?.ledger ?? 0); + const txHash = String(rawEvent?.txHash ?? rawEvent?.transactionHash ?? rawEvent?.tx_hash ?? ""); + const eventIndex = Number( + rawEvent?.eventIndex ?? rawEvent?.event_index ?? rawEvent?.eventIndex ?? fallbackIndex + ); + return `${ledger}:${txHash}:${eventIndex}`; +} + +function sortEventsByLedgerAndIndex(events) { + return [...events].sort((a, b) => { + const ledgerDelta = Number(a.ledger ?? 0) - Number(b.ledger ?? 0); + if (ledgerDelta !== 0) return ledgerDelta; + const eventDelta = Number(a.eventIndex ?? 0) - Number(b.eventIndex ?? 0); + if (eventDelta !== 0) return eventDelta; + return String(a.txHash ?? "").localeCompare(String(b.txHash ?? "")); + }); +} + function decodeEvent(rawEvent, index) { const topics = Array.isArray(rawEvent.topic) ? rawEvent.topic.map(parseVal) @@ -234,7 +253,8 @@ function decodeEvent(rawEvent, index) { const topicName = String(topics[0] ?? "").toLowerCase(); const ledger = rawEvent.ledger; - const id = rawEvent.id || `${ledger}:${rawEvent.txHash || ""}:${index}`; + const eventIndex = Number(rawEvent.eventIndex ?? rawEvent.event_index ?? index ?? 0); + const id = rawEvent.id || canonicalEventIdentity(rawEvent, index); return { id, @@ -243,6 +263,7 @@ function decodeEvent(rawEvent, index) { value, ledger, txHash: rawEvent.txHash, + eventIndex, ledgerClosedAt: rawEvent.ledgerClosedAt, }; } @@ -371,7 +392,7 @@ async function main() { limit: 100, }); - const rawEvents = res.events || []; + const rawEvents = sortEventsByLedgerAndIndex(res.events || []); const latestSeen = res.latestLedger || currentLedger; log(`[INFO] Ingesting ledger range ${currentLedger}..${latestSeen} (found ${rawEvents.length} events)`); @@ -387,13 +408,16 @@ async function main() { for (let i = 0; i < rawEvents.length; i++) { const decoded = decodeEvent(rawEvents[i], i); - if (!existingEventIds.has(decoded.id)) { + const eventKey = canonicalEventIdentity(rawEvents[i], i); + if (!existingEventIds.has(decoded.id) && !existingEventIds.has(eventKey)) { existingEventIds.add(decoded.id); + existingEventIds.add(eventKey); store.events.push(decoded); } applyEventToProposals(decoded, proposalsMap); } + store.events = sortEventsByLedgerAndIndex(store.events); store.proposals = [...proposalsMap.values()]; }