Skip to content

Commit 443956a

Browse files
ogazboizogazboiz
authored
Merge pull request #38 from MarvyNwaokobia/fix/issue-10-loan-check-cron-dedup
fix(cron): prevent duplicate repayment_due notifications per loan
2 parents 9526853 + 8d589df commit 443956a

9 files changed

Lines changed: 343 additions & 122 deletions

‎migrations/1772000000000_webhook-subscriptions.js‎

Lines changed: 5 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -11,7 +11,11 @@ export const up = (pgm) => {
1111
pgm.createTable("webhook_subscriptions", {
1212
id: "id",
1313
callback_url: { type: "text", notNull: true },
14-
event_types: { type: "jsonb", notNull: true, default: "[]::jsonb" },
14+
event_types: {
15+
type: "jsonb",
16+
notNull: true,
17+
default: pgm.func("'[]'::jsonb"),
18+
},
1519
secret: { type: "varchar(255)" },
1620
is_active: { type: "boolean", notNull: true, default: true },
1721
created_at: {

‎migrations/1778000000009_transaction-submissions.js‎

Lines changed: 14 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,7 +1,7 @@
11
/**
22
* @param { import("node-pg-migrate").MigrationBuilder } @param pgm {import("node-pg-migrate").MigrationBuilder}
33
*/
4-
exports.up = (pgm) => {
4+
export const up = (pgm) => {
55
pgm.createTable("transaction_submissions", {
66
id: {
77
type: "serial",
@@ -52,6 +52,18 @@ exports.up = (pgm) => {
5252
pgm.createIndex("transaction_submissions", ["status"]);
5353
pgm.createIndex("transaction_submissions", ["transaction_type"]);
5454

55+
// Ensure the shared updated_at trigger function exists (not created by any
56+
// earlier migration), otherwise the trigger below cannot be created.
57+
pgm.sql(`
58+
CREATE OR REPLACE FUNCTION update_updated_at_column()
59+
RETURNS trigger AS $$
60+
BEGIN
61+
NEW.updated_at = NOW();
62+
RETURN NEW;
63+
END;
64+
$$ LANGUAGE plpgsql;
65+
`);
66+
5567
// Trigger to update updated_at timestamp
5668
pgm.createTrigger("transaction_submissions", "update_updated_at", {
5769
when: "BEFORE",
@@ -63,6 +75,6 @@ exports.up = (pgm) => {
6375
/**
6476
* @param { import("node-pg-migrate").MigrationBuilder } @param pgm {import("node-pg-migrate").MigrationBuilder}
6577
*/
66-
exports.down = (pgm) => {
78+
export const down = (pgm) => {
6779
pgm.dropTable("transaction_submissions");
6880
};

‎migrations/1781000000011_webhook-retry-logic.js‎

Lines changed: 22 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -8,21 +8,31 @@ export const shorthands = undefined;
88
* @returns {Promise<void> | void}
99
*/
1010
export const up = (pgm) => {
11-
// Add payload column to webhook_deliveries table
12-
pgm.addColumn("webhook_deliveries", {
13-
payload: {
14-
type: "jsonb",
15-
notNull: false,
11+
// Add payload column to webhook_deliveries table.
12+
// 1772 already creates this column, so guard against re-adding it.
13+
pgm.addColumn(
14+
"webhook_deliveries",
15+
{
16+
payload: {
17+
type: "jsonb",
18+
notNull: false,
19+
},
1620
},
17-
});
21+
{ ifNotExists: true },
22+
);
1823

19-
// Add next_retry_at column to track when to retry
20-
pgm.addColumn("webhook_deliveries", {
21-
next_retry_at: {
22-
type: "timestamp",
23-
notNull: false,
24+
// Add next_retry_at column to track when to retry.
25+
// 1772 already creates this column, so guard against re-adding it.
26+
pgm.addColumn(
27+
"webhook_deliveries",
28+
{
29+
next_retry_at: {
30+
type: "timestamp",
31+
notNull: false,
32+
},
2433
},
25-
});
34+
{ ifNotExists: true },
35+
);
2636

2737
// Add index for efficient retry polling
2838
pgm.createIndex("webhook_deliveries", ["next_retry_at"], {

‎migrations/1784000000014_add-loan-disputes.js‎

Lines changed: 36 additions & 35 deletions
Original file line numberDiff line numberDiff line change
@@ -1,41 +1,42 @@
11
// Migration: Add loan_disputes table and support for disputed loan status
22

3-
module.exports = {
4-
async up(db) {
5-
// 1. Create loan_disputes table
6-
await db.query(`
7-
CREATE TABLE IF NOT EXISTS loan_disputes (
8-
id SERIAL PRIMARY KEY,
9-
loan_id INTEGER NOT NULL REFERENCES loan_events(loan_id),
10-
borrower TEXT NOT NULL,
11-
reason TEXT NOT NULL,
12-
status TEXT NOT NULL DEFAULT 'open', -- open, resolved, rejected
13-
admin_note TEXT,
14-
resolution TEXT,
15-
created_at TIMESTAMP WITH TIME ZONE DEFAULT NOW(),
16-
resolved_at TIMESTAMP WITH TIME ZONE
17-
);
18-
`);
3+
export const up = (pgm) => {
4+
// 1. Create loan_disputes table
5+
pgm.sql(`
6+
CREATE TABLE IF NOT EXISTS loan_disputes (
7+
id SERIAL PRIMARY KEY,
8+
-- loan_id is not a FK: loan_events is an append-only event table whose
9+
-- loan_id is non-unique (and later becomes a view), so it cannot be a
10+
-- foreign key target.
11+
loan_id INTEGER NOT NULL,
12+
borrower TEXT NOT NULL,
13+
reason TEXT NOT NULL,
14+
status TEXT NOT NULL DEFAULT 'open', -- open, resolved, rejected
15+
admin_note TEXT,
16+
resolution TEXT,
17+
created_at TIMESTAMP WITH TIME ZONE DEFAULT NOW(),
18+
resolved_at TIMESTAMP WITH TIME ZONE
19+
);
20+
`);
1921

20-
// 2. Add indexes for efficient querying
21-
await db.query(`
22-
CREATE INDEX IF NOT EXISTS idx_loan_disputes_status ON loan_disputes(status);
23-
`);
24-
await db.query(`
25-
CREATE INDEX IF NOT EXISTS idx_loan_disputes_borrower ON loan_disputes(borrower);
26-
`);
27-
await db.query(`
28-
CREATE INDEX IF NOT EXISTS idx_loan_disputes_loan_id ON loan_disputes(loan_id);
29-
`);
22+
// 2. Add indexes for efficient querying
23+
pgm.sql(`
24+
CREATE INDEX IF NOT EXISTS idx_loan_disputes_status ON loan_disputes(status);
25+
`);
26+
pgm.sql(`
27+
CREATE INDEX IF NOT EXISTS idx_loan_disputes_borrower ON loan_disputes(borrower);
28+
`);
29+
pgm.sql(`
30+
CREATE INDEX IF NOT EXISTS idx_loan_disputes_loan_id ON loan_disputes(loan_id);
31+
`);
3032

31-
// 3. Add disputed status to loan_events (if using status enum, update it)
32-
// If status is a string, no migration needed. If enum, alter type here.
33-
// Example for enum:
34-
// await db.query(`ALTER TYPE loan_status_enum ADD VALUE IF NOT EXISTS 'disputed';`);
35-
},
33+
// 3. Add disputed status to loan_events (if using status enum, update it)
34+
// If status is a string, no migration needed. If enum, alter type here.
35+
// Example for enum:
36+
// pgm.sql(`ALTER TYPE loan_status_enum ADD VALUE IF NOT EXISTS 'disputed';`);
37+
};
3638

37-
async down(db) {
38-
await db.query(`DROP TABLE IF EXISTS loan_disputes;`);
39-
// No need to remove enum value (Postgres doesn't support removing enum values easily)
40-
},
39+
export const down = (pgm) => {
40+
pgm.sql(`DROP TABLE IF EXISTS loan_disputes;`);
41+
// No need to remove enum value (Postgres doesn't support removing enum values easily)
4142
};

‎migrations/1787000000017_user-notification-preferences.js‎

Lines changed: 10 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -8,11 +8,16 @@ export const shorthands = undefined;
88
* @returns {Promise<void> | void}
99
*/
1010
export const up = (pgm) => {
11-
pgm.addColumns("user_profiles", {
12-
email_enabled: { type: "boolean", notNull: true, default: false },
13-
sms_enabled: { type: "boolean", notNull: true, default: false },
14-
phone: { type: "varchar(20)" },
15-
});
11+
// 1773 already adds these notification columns, so guard against re-adding.
12+
pgm.addColumns(
13+
"user_profiles",
14+
{
15+
email_enabled: { type: "boolean", notNull: true, default: false },
16+
sms_enabled: { type: "boolean", notNull: true, default: false },
17+
phone: { type: "varchar(20)" },
18+
},
19+
{ ifNotExists: true },
20+
);
1621
};
1722

1823
/**

‎migrations/1788000000019_unified-contract-events.js‎

Lines changed: 17 additions & 52 deletions
Original file line numberDiff line numberDiff line change
@@ -12,32 +12,15 @@ export const up = (pgm) => {
1212
// 3. Make address nullable (for events like YieldDistributed that may not have a user address)
1313
pgm.alterColumn("contract_events", "address", { notNull: false });
1414

15-
// 4. Rename indexes to match the new table and column names
16-
pgm.renameIndex(
17-
"contract_events",
18-
"idx_loan_events_borrower_event_type",
19-
"idx_contract_events_address_event_type",
20-
);
21-
pgm.renameIndex(
22-
"contract_events",
23-
"idx_loan_events_loan_id_event_type",
24-
"idx_contract_events_loan_id_event_type",
25-
);
26-
pgm.renameIndex(
27-
"contract_events",
28-
"idx_loan_events_event_type_loan_id",
29-
"idx_contract_events_event_type_loan_id",
30-
);
31-
pgm.renameIndex(
32-
"contract_events",
33-
"idx_loan_events_ledger",
34-
"idx_contract_events_ledger",
35-
);
36-
pgm.renameIndex(
37-
"contract_events",
38-
"idx_loan_events_pool_deposits_withdraws",
39-
"idx_contract_events_pool_deposits_withdraws",
40-
);
15+
// 4. Rename indexes to match the new table and column names.
16+
// node-pg-migrate has no renameIndex helper, so use raw ALTER INDEX.
17+
pgm.sql(`
18+
ALTER INDEX IF EXISTS idx_loan_events_borrower_event_type RENAME TO idx_contract_events_address_event_type;
19+
ALTER INDEX IF EXISTS idx_loan_events_loan_id_event_type RENAME TO idx_contract_events_loan_id_event_type;
20+
ALTER INDEX IF EXISTS idx_loan_events_event_type_loan_id RENAME TO idx_contract_events_event_type_loan_id;
21+
ALTER INDEX IF EXISTS idx_loan_events_ledger RENAME TO idx_contract_events_ledger;
22+
ALTER INDEX IF EXISTS idx_loan_events_pool_deposits_withdraws RENAME TO idx_contract_events_pool_deposits_withdraws;
23+
`);
4124

4225
// Rename single-column indexes from initial schema (if they exist)
4326
pgm.sql(`
@@ -81,32 +64,14 @@ export const down = (pgm) => {
8164

8265
pgm.renameTable("contract_events", "loan_events");
8366

84-
// Revert index names
85-
pgm.renameIndex(
86-
"loan_events",
87-
"idx_contract_events_address_event_type",
88-
"idx_loan_events_borrower_event_type",
89-
);
90-
pgm.renameIndex(
91-
"loan_events",
92-
"idx_contract_events_loan_id_event_type",
93-
"idx_loan_events_loan_id_event_type",
94-
);
95-
pgm.renameIndex(
96-
"loan_events",
97-
"idx_contract_events_event_type_loan_id",
98-
"idx_loan_events_event_type_loan_id",
99-
);
100-
pgm.renameIndex(
101-
"loan_events",
102-
"idx_contract_events_ledger",
103-
"idx_loan_events_ledger",
104-
);
105-
pgm.renameIndex(
106-
"loan_events",
107-
"idx_contract_events_pool_deposits_withdraws",
108-
"idx_loan_events_pool_deposits_withdraws",
109-
);
67+
// Revert index names (raw ALTER INDEX; no renameIndex helper exists)
68+
pgm.sql(`
69+
ALTER INDEX IF EXISTS idx_contract_events_address_event_type RENAME TO idx_loan_events_borrower_event_type;
70+
ALTER INDEX IF EXISTS idx_contract_events_loan_id_event_type RENAME TO idx_loan_events_loan_id_event_type;
71+
ALTER INDEX IF EXISTS idx_contract_events_event_type_loan_id RENAME TO idx_loan_events_event_type_loan_id;
72+
ALTER INDEX IF EXISTS idx_contract_events_ledger RENAME TO idx_loan_events_ledger;
73+
ALTER INDEX IF EXISTS idx_contract_events_pool_deposits_withdraws RENAME TO idx_loan_events_pool_deposits_withdraws;
74+
`);
11075

11176
pgm.sql(`
11277
ALTER INDEX IF EXISTS contract_events_event_type_index RENAME TO loan_events_event_type_index;

‎migrations/1789000000000_ensure-core-tables.js‎

Lines changed: 4 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -25,11 +25,13 @@ export const up = (pgm) => {
2525
END $$;
2626
`);
2727

28-
// Ensure loan_events table matches requested schema
28+
// Ensure loan_events relation exists. Use to_regclass (not pg_tables) so the
29+
// backward-compat loan_events VIEW created in 1788 also counts as existing;
30+
// otherwise this would try to CREATE TABLE over the view and fail.
2931
pgm.sql(`
3032
DO $$
3133
BEGIN
32-
IF NOT EXISTS (SELECT FROM pg_tables WHERE schemaname = 'public' AND tablename = 'loan_events') THEN
34+
IF to_regclass('public.loan_events') IS NULL THEN
3335
CREATE TABLE loan_events (
3436
id SERIAL PRIMARY KEY,
3537
loan_id INTEGER,

‎src/cron/loanCheckCron.ts‎

Lines changed: 43 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -7,6 +7,14 @@ import { cacheService } from "../services/cacheService.js";
77
const LOCK_KEY = "loan_due_check_cron:running";
88
const LOCK_TTL_SECONDS = 300; // 5 minutes
99

10+
const LEDGER_CLOSE_SECONDS = 5;
11+
const DEFAULT_TERM_LEDGERS = 17280; // 1 day in ledgers
12+
const NOTIFICATION_WINDOW_SECONDS = 24 * 60 * 60; // 24 hours
13+
14+
function notificationCacheKey(loanId: number): string {
15+
return `loan_due_notified:${loanId}`;
16+
}
17+
1018
export async function runLoanDueCheck(): Promise<void> {
1119
let lockAcquired = false;
1220
try {
@@ -30,31 +38,53 @@ export async function runLoanDueCheck(): Promise<void> {
3038
try {
3139
logger.info("Running loan due check cron...");
3240

33-
// Find loans where a repayment is due in the next 24 hours
34-
// This is a simplified query; in a real app, you'd check against a repayment schedule table
3541
const result = await query(`
36-
SELECT le.loan_id, le.address, le.amount
42+
SELECT le.loan_id, le.address, le.amount,
43+
le.ledger_closed_at AS approved_at,
44+
COALESCE(le.term_ledgers, ${DEFAULT_TERM_LEDGERS}) AS term_ledgers
3745
FROM contract_events le
3846
WHERE le.event_type = 'LoanApproved'
3947
AND NOT EXISTS (
40-
SELECT 1 FROM contract_events re
48+
SELECT 1 FROM contract_events re
4149
WHERE re.loan_id = le.loan_id AND re.event_type = 'LoanRepaid'
4250
)
43-
AND le.ledger_closed_at < NOW() - INTERVAL '30 days' -- Simplified due logic
51+
AND (le.ledger_closed_at + (COALESCE(le.term_ledgers, ${DEFAULT_TERM_LEDGERS}) * ${LEDGER_CLOSE_SECONDS} || ' seconds')::interval) <= NOW() + INTERVAL '24 hours'
4452
`);
4553

54+
let notifiedCount = 0;
55+
4656
for (const loan of result.rows) {
47-
await notificationService.createNotification({
48-
userId: loan.address,
49-
type: "repayment_due",
50-
title: "Repayment Due Soon",
51-
message: `Your repayment for loan #${loan.loan_id} of ${loan.amount} is due.`,
52-
loanId: loan.loan_id,
53-
});
57+
const cacheKey = notificationCacheKey(loan.loan_id);
58+
const alreadyNotified = await cacheService.setNotExists(
59+
cacheKey,
60+
"1",
61+
NOTIFICATION_WINDOW_SECONDS,
62+
);
63+
64+
if (!alreadyNotified) {
65+
continue;
66+
}
67+
68+
try {
69+
await notificationService.createNotification({
70+
userId: loan.address,
71+
type: "repayment_due",
72+
title: "Repayment Due Soon",
73+
message: `Your repayment for loan #${loan.loan_id} of ${loan.amount} is due.`,
74+
loanId: loan.loan_id,
75+
});
76+
notifiedCount++;
77+
} catch (err) {
78+
logger.error("Failed to send notification, clearing dedup key", {
79+
loanId: loan.loan_id,
80+
error: err,
81+
});
82+
await cacheService.delete(cacheKey).catch(() => {});
83+
}
5484
}
5585

5686
logger.info(
57-
`Loan due check completed. Notified ${result.rows.length} borrowers.`,
87+
`Loan due check completed. Notified ${notifiedCount} borrowers (${result.rows.length} due loans found).`,
5888
);
5989
} catch (error) {
6090
logger.error("Error in loan due check cron", { error });

0 commit comments

Comments
 (0)