From c6a559411d8f1737aadf7cb255d605619c345509 Mon Sep 17 00:00:00 2001 From: peacebas Date: Sun, 18 Oct 2026 19:44:16 +0100 Subject: [PATCH] feat: fix #814 stats routes, #824 webhook deliveries, #826 redis scan, #834 jest open handles --- .github/workflows/ci.yml | 14 + docs/openapi.yaml | 249 ++++++++++++++ jest.config.js | 2 + src/__tests__/CacheService.test.ts | 119 +++++++ src/__tests__/SongArtistStatsRoutes.test.ts | 266 +++++++++++++++ src/__tests__/WebhookDeliveriesRoutes.test.ts | 283 ++++++++++++++++ src/config/db.ts | 1 + src/config/redis.ts | 4 + src/controllers/SongController.ts | 102 +++++- src/controllers/WebhookController.ts | 88 ++++- src/entities/WebhookDelivery.ts | 67 ++++ src/entities/WebhookSubscription.ts | 4 + src/index.ts | 19 +- src/jobs/WebhookRetentionJob.ts | 34 ++ ...-AddWebhookDeliveriesAndFailureTracking.ts | 144 ++++++++ src/routes/SongRoutes.ts | 6 +- src/routes/artistRoutes.ts | 10 +- src/routes/webhookRoutes.ts | 2 + src/services/AccountDeletionService.ts | 13 +- src/services/CacheService.ts | 32 +- src/services/DatabaseConnectionManager.ts | 26 +- src/services/WebhookService.ts | 311 ++++++++++++++++-- .../__tests__/AccountDeletionService.test.ts | 17 +- src/services/privy/PrivySessionService.ts | 20 +- tests/setup/jestTeardown.ts | 46 +++ 25 files changed, 1809 insertions(+), 70 deletions(-) create mode 100644 src/__tests__/CacheService.test.ts create mode 100644 src/__tests__/SongArtistStatsRoutes.test.ts create mode 100644 src/__tests__/WebhookDeliveriesRoutes.test.ts create mode 100644 src/entities/WebhookDelivery.ts create mode 100644 src/jobs/WebhookRetentionJob.ts create mode 100644 src/migrations/1757700000000-AddWebhookDeliveriesAndFailureTracking.ts create mode 100644 tests/setup/jestTeardown.ts diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index e51268b3..7df7c79b 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -25,6 +25,20 @@ jobs: - name: Run ESLint run: npm run lint + test: + name: Unit & Integration tests + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + - name: Setup Node.js + uses: actions/setup-node@v4 + with: + node-version: 20 + - name: Install dependencies + run: npm ci + - name: Run tests with --detectOpenHandles + run: npm test -- --detectOpenHandles + audit: name: Dependency vulnerability gate runs-on: ubuntu-latest diff --git a/docs/openapi.yaml b/docs/openapi.yaml index a41f1a37..388e02ea 100644 --- a/docs/openapi.yaml +++ b/docs/openapi.yaml @@ -90,6 +90,189 @@ paths: $ref: '#/components/schemas/IndexedEvent' pagination: $ref: '#/components/schemas/Pagination' + /api/song/{id}/stats: + get: + summary: Get song statistics + description: Returns aggregated playback and engagement metrics for a song. Public callers receive total plays and saves; the song artist or admins receive full metrics including unique listeners and revenue. + parameters: + - name: id + in: path + required: true + description: Unique song identifier (UUID). + schema: + type: string + format: uuid + - name: window + in: query + required: false + description: Aggregation time window. Default is 7d. + schema: + type: string + enum: [24h, 7d, 30d, 90d, all-time] + default: 7d + responses: + '200': + description: Song statistics retrieved successfully. + headers: + Cache-Control: + schema: + type: string + description: Caching policy (public or private with 300s TTL). + Vary: + schema: + type: string + example: Authorization + content: + application/json: + schema: + type: object + properties: + success: + type: boolean + example: true + data: + $ref: '#/components/schemas/SongStats' + '400': + description: Invalid window parameter. + '404': + description: Song not found. + + /api/song/artist/{artistId}/stats: + get: + summary: Get artist aggregated statistics via song route + description: Returns catalog-wide metrics for the specified artist. Public callers receive plays and saves; the artist owner or admins receive full metrics including revenue and unique listeners. + parameters: + - name: artistId + in: path + required: true + description: Unique artist identifier (UUID). + schema: + type: string + format: uuid + - name: window + in: query + required: false + description: Aggregation time window. Default is 7d. + schema: + type: string + enum: [24h, 7d, 30d, 90d, all-time] + default: 7d + responses: + '200': + description: Artist statistics retrieved successfully. + '400': + description: Invalid window parameter. + '404': + description: Artist not found. + + /api/artist/{id}/stats: + get: + summary: Get artist aggregated statistics + description: Returns catalog-wide metrics for the specified artist. Public callers receive plays and saves; the artist owner or admins receive full metrics including revenue and unique listeners. + parameters: + - name: id + in: path + required: true + description: Unique artist identifier (UUID). + schema: + type: string + format: uuid + - name: window + in: query + required: false + description: Aggregation time window. Default is 7d. + schema: + type: string + enum: [24h, 7d, 30d, 90d, all-time] + default: 7d + responses: + '200': + description: Artist statistics retrieved successfully. + '400': + description: Invalid window parameter. + '404': + description: Artist not found. + + /api/webhooks/{id}/deliveries: + get: + summary: List webhook delivery attempts + description: Returns paginated delivery history for a webhook subscription (newest first). Only accessible by the subscription owner. + parameters: + - name: id + in: path + required: true + description: Unique webhook subscription identifier (UUID). + schema: + type: string + format: uuid + - name: page + in: query + required: false + schema: + type: integer + default: 1 + - name: limit + in: query + required: false + schema: + type: integer + default: 20 + responses: + '200': + description: Delivery attempts list. + content: + application/json: + schema: + type: object + properties: + success: + type: boolean + example: true + data: + type: array + items: + $ref: '#/components/schemas/WebhookDelivery' + pagination: + $ref: '#/components/schemas/Pagination' + '401': + description: Unauthorized. + '403': + description: Forbidden - caller is not the subscription owner. + '404': + description: Webhook subscription not found. + + /api/webhooks/{id}/deliveries/{deliveryId}/redeliver: + post: + summary: Redeliver a webhook attempt + description: Resends the event payload of a previous delivery attempt. Only accessible by the subscription owner. + parameters: + - name: id + in: path + required: true + description: Webhook subscription UUID. + schema: + type: string + format: uuid + - name: deliveryId + in: path + required: true + description: Webhook delivery attempt UUID. + schema: + type: string + format: uuid + responses: + '200': + description: Webhook redelivered successfully. + '400': + description: No payload stored for this delivery. + '401': + description: Unauthorized. + '403': + description: Forbidden - caller is not the subscription owner. + '404': + description: Subscription or delivery record not found. + '502': + description: Destination endpoint failed after retry attempts. components: schemas: IndexedEvent: @@ -130,6 +313,71 @@ components: updatedAt: type: string format: date-time + SongStats: + type: object + properties: + window: + type: string + example: 7d + from: + type: string + format: date-time + to: + type: string + format: date-time + totals: + type: object + properties: + plays: + type: integer + example: 120 + saves: + type: integer + example: 30 + uniqueListeners: + type: integer + description: Owner and admin only. + example: 45 + revenueStroops: + type: string + description: Owner and admin only. + example: "5000000" + WebhookDelivery: + type: object + properties: + id: + type: string + format: uuid + subscriptionId: + type: string + format: uuid + eventId: + type: string + eventType: + type: string + attempt: + type: integer + example: 1 + status: + type: string + enum: [pending, succeeded, failed] + responseStatus: + type: integer + nullable: true + example: 200 + durationMs: + type: integer + nullable: true + example: 45 + error: + type: string + nullable: true + payload: + type: object + nullable: true + createdAt: + type: string + format: date-time Pagination: type: object properties: @@ -145,3 +393,4 @@ components: totalPages: type: integer example: 3 + diff --git a/jest.config.js b/jest.config.js index affe4ba1..fdb4cc48 100644 --- a/jest.config.js +++ b/jest.config.js @@ -7,6 +7,8 @@ module.exports = { // Strip AI vendor keys and pin AI_PROVIDER=noop before every test file so // no test can depend on a live AI provider (#287). setupFiles: ["/tests/setup/aiProviderIsolation.js"], + // Global teardown to clean up in-process timers and mock connections (Issue #834) + setupFilesAfterEnv: ["/tests/setup/jestTeardown.ts"], moduleNameMapper: { "^reflect-metadata$": "/node_modules/reflect-metadata", // wasm-attestation-bindings is ESM-only and uses `import.meta`, which diff --git a/src/__tests__/CacheService.test.ts b/src/__tests__/CacheService.test.ts new file mode 100644 index 00000000..258608ba --- /dev/null +++ b/src/__tests__/CacheService.test.ts @@ -0,0 +1,119 @@ +import { Readable } from 'stream'; +import redis from '../config/redis'; +import { CacheService } from '../services/CacheService'; + +jest.mock('../config/redis', () => ({ + __esModule: true, + default: { + get: jest.fn(), + set: jest.fn(), + del: jest.fn(), + unlink: jest.fn(), + scanStream: jest.fn(), + }, +})); + +jest.mock('../config/logger', () => ({ + __esModule: true, + default: { + info: jest.fn(), + warn: jest.fn(), + error: jest.fn(), + debug: jest.fn(), + }, +})); + +describe('CacheService', () => { + beforeEach(() => { + jest.clearAllMocks(); + }); + + describe('invalidatePattern (Issue #826)', () => { + it('uses scanStream with count 500 and non-blocking unlink instead of keys/del', async () => { + const chunk1 = ['stats:song:s1:30d', 'stats:song:s1:7d']; + const chunk2 = ['stats:song:s1:all-time']; + + // Mock an async iterable stream like ioredis scanStream + const mockStream = Readable.from([chunk1, chunk2]); + (redis.scanStream as jest.Mock).mockReturnValue(mockStream); + (redis.unlink as jest.Mock).mockResolvedValue(chunk1.length); + + await CacheService.invalidatePattern('stats:song:s1:*'); + + expect(redis.scanStream).toHaveBeenCalledWith({ + match: 'stats:song:s1:*', + count: 500, + }); + + expect(redis.unlink).toHaveBeenCalledWith(...chunk1); + expect(redis.unlink).toHaveBeenCalledWith(...chunk2); + }); + + it('falls back to redis.del if redis.unlink is not available', async () => { + const keys = ['song:123']; + const mockStream = Readable.from([keys]); + (redis.scanStream as jest.Mock).mockReturnValue(mockStream); + const originalUnlink = redis.unlink; + (redis as any).unlink = undefined; + (redis.del as jest.Mock).mockResolvedValue(1); + + await CacheService.invalidatePattern('song:123'); + + expect(redis.del).toHaveBeenCalledWith(...keys); + + (redis as any).unlink = originalUnlink; + }); + + it('handles empty scanStream gracefully without deleting anything', async () => { + const mockStream = Readable.from([]); + (redis.scanStream as jest.Mock).mockReturnValue(mockStream); + + await CacheService.invalidatePattern('nonexistent:*'); + + expect(redis.scanStream).toHaveBeenCalledWith({ + match: 'nonexistent:*', + count: 500, + }); + expect(redis.unlink).not.toHaveBeenCalled(); + expect(redis.del).not.toHaveBeenCalled(); + }); + + it('catches and logs errors without throwing', async () => { + (redis.scanStream as jest.Mock).mockImplementation(() => { + throw new Error('Redis connection failure'); + }); + + await expect(CacheService.invalidatePattern('error:*')).resolves.not.toThrow(); + }); + }); + + describe('basic get, set, and invalidate', () => { + it('sets cache value with TTL in seconds', async () => { + (redis.set as jest.Mock).mockResolvedValue('OK'); + await CacheService.set('test:key', { foo: 'bar' }, 60000); + + expect(redis.set).toHaveBeenCalledWith('test:key', JSON.stringify({ foo: 'bar' }), 'EX', 60); + }); + + it('gets and parses JSON from cache', async () => { + (redis.get as jest.Mock).mockResolvedValue(JSON.stringify({ hello: 'world' })); + const result = await CacheService.get<{ hello: string }>('test:key'); + + expect(result).toEqual({ hello: 'world' }); + }); + + it('invalidates a single key', async () => { + (redis.del as jest.Mock).mockResolvedValue(1); + await CacheService.invalidate('test:key'); + + expect(redis.del).toHaveBeenCalledWith('test:key'); + }); + + it('clears song cache using clearSong', async () => { + (redis.del as jest.Mock).mockResolvedValue(1); + await CacheService.clearSong('song-abc'); + + expect(redis.del).toHaveBeenCalledWith('song:song-abc'); + }); + }); +}); diff --git a/src/__tests__/SongArtistStatsRoutes.test.ts b/src/__tests__/SongArtistStatsRoutes.test.ts new file mode 100644 index 00000000..4761882b --- /dev/null +++ b/src/__tests__/SongArtistStatsRoutes.test.ts @@ -0,0 +1,266 @@ +import 'reflect-metadata'; +import request from 'supertest'; +import express, { Express } from 'express'; +import songRoutes from '../routes/SongRoutes'; +import artistRoutes from '../routes/artistRoutes'; +import AppDataSource from '../config/db'; +import { SongController } from '../controllers/SongController'; + +// Mock SongStatsService +const mockFullStats = { + window: '7d', + from: '2026-10-11T00:00:00.000Z', + to: '2026-10-18T00:00:00.000Z', + totals: { + plays: 120, + uniqueListeners: 45, + saves: 30, + revenueStroops: '5000000', + }, + previous: { + plays: 100, + uniqueListeners: 40, + saves: 25, + revenueStroops: '4000000', + }, + comparison: { + playsPct: 20, + uniqueListenersPct: 12.5, + savesPct: 20, + revenuePct: 25, + }, + series: [ + { bucket: '2026-10-11', plays: 10, saves: 2, revenueStroops: '500000' }, + ], +}; + +const mockSong = { + id: 'song-uuid-1', + title: 'Stellar Track', + artistId: 'artist-uuid-1', +}; + +const mockArtist = { + id: 'artist-uuid-1', + name: 'Stellar Artist', +}; + +jest.mock('../config/db', () => ({ + __esModule: true, + default: { + getRepository: jest.fn(), + }, +})); + +jest.mock('../config/redis', () => ({ + __esModule: true, + default: { + get: jest.fn(), + set: jest.fn(), + unlink: jest.fn(), + }, +})); + +jest.mock('../services/Song/SongStatsService', () => { + return { + SongStatsService: jest.fn().mockImplementation(() => ({ + getSongStats: jest.fn(async (songId: string, _window: string) => { + if (songId === 'song-uuid-1') return mockFullStats; + const err: any = new Error('Song not found'); + err.statusCode = 404; + throw err; + }), + getArtistStats: jest.fn(async (artistId: string, _window: string) => { + if (artistId === 'artist-uuid-1') return mockFullStats; + const err: any = new Error('Artist not found'); + err.statusCode = 404; + throw err; + }), + })), + parseWindow: jest.fn((w?: string) => { + const allowed = ['24h', '7d', '30d', '90d', 'all-time']; + const target = (w || '7d').toLowerCase(); + if (!allowed.includes(target)) { + throw new Error(`Invalid window: "${w}". Expected one of: ${allowed.join(', ')}`); + } + return target; + }), + }; +}); + +describe('Song and Artist Stats Endpoints (Issue #814)', () => { + let app: Express; + let authUser: any = null; + + beforeAll(() => { + // Setup Express application + app = express(); + app.use(express.json()); + + // Inject caller into request for auth simulation + app.use((req, _res, next) => { + if (authUser) { + (req as any).user = authUser; + } + next(); + }); + + app.use('/api/song', songRoutes); + app.use('/api/artist', artistRoutes); + + // Mock typeorm getRepository for Song and User + (AppDataSource.getRepository as jest.Mock).mockImplementation((entity: any) => { + const name = entity?.name; + if (name === 'Song') { + return { + findOne: jest.fn(async ({ where }: any) => { + if (where?.id === 'song-uuid-1') return mockSong; + return null; + }), + }; + } + if (name === 'User') { + return { + findOne: jest.fn(async ({ where }: any) => { + if (where?.id === 'artist-uuid-1') return mockArtist; + return null; + }), + }; + } + return {}; + }); + }); + + beforeEach(() => { + authUser = null; + }); + + describe('GET /api/song/:id/stats', () => { + it('returns only public metrics for unauthenticated users', async () => { + const res = await request(app) + .get('/api/song/song-uuid-1/stats') + .expect(200); + + expect(res.body.success).toBe(true); + expect(res.body.data.totals.plays).toBe(120); + expect(res.body.data.totals.saves).toBe(30); + + // Owner-only fields MUST NOT be present + expect(res.body.data.totals.uniqueListeners).toBeUndefined(); + expect(res.body.data.totals.revenueStroops).toBeUndefined(); + if (res.body.data.previous) { + expect(res.body.data.previous.uniqueListeners).toBeUndefined(); + expect(res.body.data.previous.revenueStroops).toBeUndefined(); + } + if (res.body.data.comparison) { + expect(res.body.data.comparison.uniqueListenersPct).toBeUndefined(); + expect(res.body.data.comparison.revenuePct).toBeUndefined(); + } + + // Check caching headers + expect(res.headers['cache-control']).toContain('public'); + expect(res.headers['cache-control']).toContain('max-age=300'); + expect(res.headers['vary']).toBe('Authorization'); + }); + + it('returns only public metrics for a non-owner authenticated user', async () => { + authUser = { id: 'other-user-uuid', role: 'listener' }; + + const res = await request(app) + .get('/api/song/song-uuid-1/stats') + .expect(200); + + expect(res.body.success).toBe(true); + expect(res.body.data.totals.plays).toBe(120); + expect(res.body.data.totals.uniqueListeners).toBeUndefined(); + expect(res.body.data.totals.revenueStroops).toBeUndefined(); + + expect(res.headers['cache-control']).toContain('public'); + }); + + it('returns complete metrics including revenue and unique listeners to song owner', async () => { + authUser = { id: 'artist-uuid-1', role: 'artist' }; + + const res = await request(app) + .get('/api/song/song-uuid-1/stats') + .expect(200); + + expect(res.body.success).toBe(true); + expect(res.body.data.totals.plays).toBe(120); + expect(res.body.data.totals.uniqueListeners).toBe(45); + expect(res.body.data.totals.revenueStroops).toBe('5000000'); + expect(res.body.data.comparison.revenuePct).toBe(25); + + // Owner gets private caching header + expect(res.headers['cache-control']).toContain('private'); + expect(res.headers['cache-control']).toContain('max-age=300'); + expect(res.headers['vary']).toBe('Authorization'); + }); + + it('returns complete metrics to an admin user', async () => { + authUser = { id: 'admin-user-uuid', role: 'admin' }; + + const res = await request(app) + .get('/api/song/song-uuid-1/stats') + .expect(200); + + expect(res.body.success).toBe(true); + expect(res.body.data.totals.uniqueListeners).toBe(45); + expect(res.body.data.totals.revenueStroops).toBe('5000000'); + }); + + it('rejects invalid time window with 400 Bad Request', async () => { + const res = await request(app) + .get('/api/song/song-uuid-1/stats?window=invalid_period') + .expect(400); + + expect(res.body.success).toBe(false); + expect(res.body.message).toContain('Invalid window'); + }); + + it('returns 404 when song does not exist', async () => { + const res = await request(app) + .get('/api/song/nonexistent-song/stats') + .expect(404); + + expect(res.body.success).toBe(false); + }); + }); + + describe('Artist stats endpoints (/api/artist/:id/stats and /api/song/artist/:artistId/stats)', () => { + it('is reachable via GET /api/artist/:id/stats and returns public counts to non-owner', async () => { + const res = await request(app) + .get('/api/artist/artist-uuid-1/stats') + .expect(200); + + expect(res.body.success).toBe(true); + expect(res.body.data.totals.plays).toBe(120); + expect(res.body.data.totals.uniqueListeners).toBeUndefined(); + expect(res.body.data.totals.revenueStroops).toBeUndefined(); + expect(res.headers['cache-control']).toContain('public'); + }); + + it('is reachable via GET /api/song/artist/:artistId/stats and returns owner metrics to artist', async () => { + authUser = { id: 'artist-uuid-1', role: 'artist' }; + + const res = await request(app) + .get('/api/song/artist/artist-uuid-1/stats') + .expect(200); + + expect(res.body.success).toBe(true); + expect(res.body.data.totals.plays).toBe(120); + expect(res.body.data.totals.uniqueListeners).toBe(45); + expect(res.body.data.totals.revenueStroops).toBe('5000000'); + expect(res.headers['cache-control']).toContain('private'); + }); + + it('rejects invalid window on artist stats endpoint', async () => { + const res = await request(app) + .get('/api/artist/artist-uuid-1/stats?window=2years') + .expect(400); + + expect(res.body.success).toBe(false); + expect(res.body.message).toContain('Invalid window'); + }); + }); +}); diff --git a/src/__tests__/WebhookDeliveriesRoutes.test.ts b/src/__tests__/WebhookDeliveriesRoutes.test.ts new file mode 100644 index 00000000..3c00a9cf --- /dev/null +++ b/src/__tests__/WebhookDeliveriesRoutes.test.ts @@ -0,0 +1,283 @@ +import 'reflect-metadata'; +import request from 'supertest'; +import express, { Express } from 'express'; +import { WebhookService } from '../services/WebhookService'; +import { WebhookDelivery, WebhookDeliveryStatus } from '../entities/WebhookDelivery'; +import { WebhookController } from '../controllers/WebhookController'; + +// Mock DB, logger, and authMiddleware +jest.mock('../config/db', () => ({ + __esModule: true, + default: { + getRepository: jest.fn(), + }, +})); + +jest.mock('../config/logger', () => ({ + __esModule: true, + default: { + info: jest.fn(), + warn: jest.fn(), + error: jest.fn(), + debug: jest.fn(), + }, +})); + +jest.mock('../middlewares/authMiddleware', () => ({ + requireAuth: (req: any, res: any, next: any) => { + if (!req.user) { + return res.status(401).json({ success: false, message: 'Unauthorized' }); + } + next(); + }, +})); + +import AppDataSource from '../config/db'; +import webhookRoutes from '../routes/webhookRoutes'; + +describe('Webhook Delivery Tracking & Redeliver (Issue #824)', () => { + let app: Express; + let authUser: any = null; + + const mockSubscription: any = { + id: 'sub-uuid-1', + userId: 'user-uuid-1', + endpoint: 'https://example.com/webhook', + secret: 'test-secret', + isActive: true, + consecutiveFailures: 0, + eventTypes: ['song.minted'], + createdAt: new Date(), + }; + + const mockDelivery: any = { + id: 'del-uuid-1', + subscriptionId: 'sub-uuid-1', + eventId: 'evt-uuid-1', + eventType: 'song.minted', + attempt: 1, + status: WebhookDeliveryStatus.SUCCEEDED, + responseStatus: 200, + durationMs: 45, + error: null, + payload: { eventId: 'evt-uuid-1', eventType: 'song.minted', songId: 'song-1' }, + createdAt: new Date(), + }; + + let mockDeliveryRepo: any; + let mockSubRepo: any; + let mockNotificationService: any; + let fetchMock: jest.Mock; + + beforeEach(() => { + authUser = { id: 'user-uuid-1', role: 'artist' }; + fetchMock = jest.fn(); + + mockDeliveryRepo = { + save: jest.fn(async (d) => ({ ...d, id: d.id || 'del-uuid-new', createdAt: new Date() })), + findAndCount: jest.fn(async () => [[mockDelivery], 1]), + findOne: jest.fn(async ({ where }) => { + if (where?.id === 'del-uuid-1' && where?.subscriptionId === 'sub-uuid-1') { + return { ...mockDelivery }; + } + return null; + }), + createQueryBuilder: jest.fn(() => ({ + delete: jest.fn(() => ({ + from: jest.fn(() => ({ + where: jest.fn(() => ({ + execute: jest.fn(async () => ({ affected: 5 })), + })), + })), + })), + })), + }; + + mockSubRepo = { + findOne: jest.fn(async ({ where }) => { + if (where?.id === 'sub-uuid-1') return { ...mockSubscription }; + return null; + }), + save: jest.fn(async (s) => s), + find: jest.fn(async () => [{ ...mockSubscription }]), + }; + + mockNotificationService = { + create: jest.fn(async (n) => n), + }; + + (AppDataSource.getRepository as jest.Mock).mockImplementation((target: any) => { + const name = target?.name; + if (name === 'WebhookDelivery') return mockDeliveryRepo; + if (name === 'WebhookSubscription') return mockSubRepo; + return {}; + }); + + const svc = new WebhookService( + fetchMock, + mockSubRepo, + mockDeliveryRepo, + mockNotificationService, + ); + WebhookController.setWebhookService(svc); + + // Setup express app + app = express(); + app.use(express.json()); + app.use((req, _res, next) => { + if (authUser) { + (req as any).user = authUser; + } + next(); + }); + app.use('/api/webhooks', webhookRoutes); + }); + + describe('WebhookService Unit Logic', () => { + it('records delivery attempt row upon successful delivery', async () => { + fetchMock.mockResolvedValueOnce({ ok: true, status: 200 }); + + const svc = new WebhookService( + fetchMock, + mockSubRepo, + mockDeliveryRepo, + mockNotificationService, + ); + + const payload = { + eventId: 'evt-test-1', + eventType: 'song.minted', + timestamp: new Date().toISOString(), + }; + + await svc.deliver('https://example.com/webhook', payload, 'secret', 'sub-uuid-1'); + + expect(fetchMock).toHaveBeenCalledTimes(1); + expect(mockDeliveryRepo.save).toHaveBeenCalledWith( + expect.objectContaining({ + subscriptionId: 'sub-uuid-1', + eventId: 'evt-test-1', + eventType: 'song.minted', + attempt: 1, + status: WebhookDeliveryStatus.SUCCEEDED, + responseStatus: 200, + }), + ); + }); + + it('records failed attempt rows and auto-disables subscription after N consecutive failed events', async () => { + fetchMock.mockResolvedValue({ ok: false, status: 500, statusText: 'Internal Server Error' }); + + // Subscription already has 4 consecutive failures + mockSubscription.consecutiveFailures = 4; + mockSubscription.isActive = true; + + const svc = new WebhookService( + fetchMock, + mockSubRepo, + mockDeliveryRepo, + mockNotificationService, + ); + + const payload = { + eventId: 'evt-fail-1', + eventType: 'song.minted', + timestamp: new Date().toISOString(), + }; + + await expect( + svc.deliver('https://example.com/webhook', payload, 'secret', 'sub-uuid-1'), + ).rejects.toThrow('Webhook delivery failed after 3 attempts'); + + // 3 attempts recorded in delivery table + expect(mockDeliveryRepo.save).toHaveBeenCalledTimes(3); + + // Subscription consecutive failures incremented to 5, isActive set to false + expect(mockSubRepo.save).toHaveBeenCalledWith( + expect.objectContaining({ + id: 'sub-uuid-1', + consecutiveFailures: 5, + isActive: false, + }), + ); + + // Owner notified via NotificationService + expect(mockNotificationService.create).toHaveBeenCalledWith( + expect.objectContaining({ + userId: 'user-uuid-1', + type: 'system', + title: 'Webhook subscription disabled', + }), + ); + }); + + it('cleans up old delivery rows older than 30 days', async () => { + const svc = new WebhookService( + fetchMock, + mockSubRepo, + mockDeliveryRepo, + mockNotificationService, + ); + + const deleted = await svc.cleanupOldDeliveries(30); + expect(deleted).toBe(5); + expect(mockDeliveryRepo.createQueryBuilder).toHaveBeenCalled(); + }); + }); + + describe('GET /api/webhooks/:id/deliveries', () => { + it('returns paginated delivery records for subscription owner', async () => { + const res = await request(app) + .get('/api/webhooks/sub-uuid-1/deliveries') + .expect(200); + + expect(res.body.success).toBe(true); + expect(res.body.data).toHaveLength(1); + expect(res.body.data[0].id).toBe('del-uuid-1'); + expect(res.body.pagination.total).toBe(1); + }); + + it('returns 403 when user is not the subscription owner', async () => { + authUser = { id: 'other-user-uuid', role: 'artist' }; + + const res = await request(app) + .get('/api/webhooks/sub-uuid-1/deliveries') + .expect(403); + + expect(res.body.success).toBe(false); + expect(res.body.message).toContain('Forbidden'); + }); + + it('rejects unauthenticated requests with 401', async () => { + authUser = null; + + const res = await request(app) + .get('/api/webhooks/sub-uuid-1/deliveries') + .expect(401); + + expect(res.body.success).toBe(false); + }); + }); + + describe('POST /api/webhooks/:id/deliveries/:deliveryId/redeliver', () => { + it('triggers redelivery of previous delivery for subscription owner', async () => { + fetchMock.mockResolvedValue({ ok: true, status: 200 }); + + const res = await request(app) + .post('/api/webhooks/sub-uuid-1/deliveries/del-uuid-1/redeliver') + .expect(200); + + expect(res.body.success).toBe(true); + expect(res.body.message).toBe('Webhook redelivered successfully'); + }); + + it('returns 404 when delivery record does not exist', async () => { + const res = await request(app) + .post('/api/webhooks/sub-uuid-1/deliveries/nonexistent-del/redeliver') + .expect(404); + + expect(res.body.success).toBe(false); + expect(res.body.message).toContain('not found'); + }); + }); +}); diff --git a/src/config/db.ts b/src/config/db.ts index 71ffbbaf..37c7549d 100644 --- a/src/config/db.ts +++ b/src/config/db.ts @@ -8,6 +8,7 @@ import { Genre } from '../entities/Genre'; import { Album } from '../entities/Album'; import { RoyaltyPayout } from '../entities/RoyaltyPayout'; import { WebhookSubscription } from '../entities/WebhookSubscription'; +import { WebhookDelivery } from '../entities/WebhookDelivery'; import { TakedownRequest } from '../entities/TakedownRequest'; import { ApiKey } from '../entities/ApiKey'; import { AiGenerationRecord } from '../entities/AiGenerationRecord'; diff --git a/src/config/redis.ts b/src/config/redis.ts index 8fd7290b..9756429f 100644 --- a/src/config/redis.ts +++ b/src/config/redis.ts @@ -1,5 +1,7 @@ import Redis from 'ioredis'; +const isTest = process.env.NODE_ENV === 'test'; + const redis = new Redis({ username: process.env.REDIS_USERNAME, host: process.env.REDIS_HOST, @@ -7,6 +9,8 @@ const redis = new Redis({ password: process.env.REDIS_PASSWORD, // tls: {}, // Required for Redis Cloud TLS maxRetriesPerRequest: null, // prevents retry errors + lazyConnect: isTest, + retryStrategy: isTest ? () => null : undefined, }); redis.on('connect', () => { diff --git a/src/controllers/SongController.ts b/src/controllers/SongController.ts index 5d983834..57ff9c84 100644 --- a/src/controllers/SongController.ts +++ b/src/controllers/SongController.ts @@ -2,6 +2,7 @@ import { Request, Response } from 'express'; import { generateListenerKey } from '../utils/listenerKey'; import AppDataSource from '../config/db'; import { Song } from '../entities/Song'; +import { UserRole } from '../entities/User'; import { SongPlayEvent } from '../entities/SongPlayEvent'; import { SongSave } from '../entities/SongSave'; import redis from '../config/redis'; @@ -351,25 +352,112 @@ export class SongController { } }; - /** GET /api/songs/:id/stats — aggregated song statistics (Issue #87). */ + /** GET /api/song/:id/stats — aggregated song statistics (Issue #814). */ static getSongStats = async (req: Request, res: Response) => { try { const songId = req.params.id as string; - const window = parseWindow(req.query.window); + let window: string; + try { + window = parseWindow(req.query.window); + } catch (err: any) { + throw AppError.validation(err.message, undefined, 'INVALID_WINDOW'); + } + + const songRepo = AppDataSource.getRepository(Song); + const song = await songRepo.findOne({ + where: { id: songId }, + select: ['id', 'artistId'], + }); + if (!song) { + throw AppError.notFound('Song not found', undefined, 'SONG_NOT_FOUND'); + } + + const caller = (req as any).user; + const isOwner = Boolean( + caller && + (caller.id === song.artistId || + caller.role === UserRole.ADMIN || + caller.role === UserRole.MODERATOR), + ); + const stats = await songStatsService.getSongStats(songId, window); - return res.status(200).json({ success: true, data: stats }); + + const data = isOwner + ? stats + : { + ...stats, + totals: { + plays: stats.totals.plays, + saves: stats.totals.saves, + }, + previous: stats.previous + ? { + plays: stats.previous.plays, + saves: stats.previous.saves, + } + : null, + comparison: stats.comparison + ? { + plays: stats.comparison.plays, + saves: stats.comparison.saves, + } + : null, + }; + + res.set('Cache-Control', isOwner ? 'private, max-age=300' : 'public, max-age=300'); + res.set('Vary', 'Authorization'); + return res.status(200).json({ success: true, data }); } catch (error) { handleError(req, res, error); } }; - /** GET /api/artists/:id/stats — artist-level aggregation (Issue #87). */ + /** GET /api/song/artist/:artistId/stats or /api/artist/:id/stats — artist-level aggregation (Issue #814). */ static getArtistStats = async (req: Request, res: Response) => { try { - const artistId = req.params.id as string; - const window = parseWindow(req.query.window); + const artistId = (req.params.artistId || req.params.id) as string; + let window: string; + try { + window = parseWindow(req.query.window); + } catch (err: any) { + throw AppError.validation(err.message, undefined, 'INVALID_WINDOW'); + } + + const caller = (req as any).user; + const isOwner = Boolean( + caller && + (caller.id === artistId || + caller.role === UserRole.ADMIN || + caller.role === UserRole.MODERATOR), + ); + const stats = await songStatsService.getArtistStats(artistId, window); - return res.status(200).json({ success: true, data: stats }); + + const data = isOwner + ? stats + : { + ...stats, + totals: { + plays: stats.totals.plays, + saves: stats.totals.saves, + }, + previous: stats.previous + ? { + plays: stats.previous.plays, + saves: stats.previous.saves, + } + : null, + comparison: stats.comparison + ? { + plays: stats.comparison.plays, + saves: stats.comparison.saves, + } + : null, + }; + + res.set('Cache-Control', isOwner ? 'private, max-age=300' : 'public, max-age=300'); + res.set('Vary', 'Authorization'); + return res.status(200).json({ success: true, data }); } catch (error) { handleError(req, res, error); } diff --git a/src/controllers/WebhookController.ts b/src/controllers/WebhookController.ts index b6e485e3..237141bb 100644 --- a/src/controllers/WebhookController.ts +++ b/src/controllers/WebhookController.ts @@ -6,10 +6,20 @@ import { User, UserRole } from '../entities/User'; import AppDataSource from '../config/db'; import logger from '../config/logger'; -const webhookService = new WebhookService(); -const userRepo: Repository = AppDataSource.getRepository(User); +let defaultWebhookService: WebhookService | null = null; +function getWebhookService(): WebhookService { + if (!defaultWebhookService) { + defaultWebhookService = new WebhookService(); + } + return defaultWebhookService; +} export class WebhookController { + public static getWebhookService = getWebhookService; + public static setWebhookService(service: WebhookService | null): void { + defaultWebhookService = service; + } + static register = async (req: Request, res: Response) => { try { const userId = (req as any).user?.id; @@ -19,7 +29,7 @@ export class WebhookController { if (!endpoint) return res.status(400).json({ success: false, message: 'endpoint is required' }); - const subscription = await webhookService.registerSubscription( + const subscription = await getWebhookService().registerSubscription( userId, endpoint, eventTypes, @@ -35,7 +45,7 @@ export class WebhookController { try { const userId = (req as any).user?.id; if (!userId) return res.status(401).json({ success: false, message: 'Unauthorized' }); - const subs = await webhookService.listSubscriptions(userId); + const subs = await getWebhookService().listSubscriptions(userId); return res.status(200).json({ success: true, data: subs }); } catch (error) { handleError(res, error); @@ -47,7 +57,7 @@ export class WebhookController { const userId = (req as any).user?.id; if (!userId) return res.status(401).json({ success: false, message: 'Unauthorized' }); const id = req.params.id as string; - await webhookService.deleteSubscription(userId, id); + await getWebhookService().deleteSubscription(userId, id); return res.status(200).json({ success: true, message: 'Webhook subscription deleted' }); } catch (error) { handleError(res, error); @@ -60,7 +70,7 @@ export class WebhookController { const userId = (req as any).user?.id; const id = req.params.id as string; // Simple test: publish a test event to this subscription only if owned - const subs = await webhookService.listSubscriptions(userId); + const subs = await getWebhookService().listSubscriptions(userId); const target = subs.find((s) => s.id === id); if (!target) return res.status(404).json({ success: false, message: 'Subscription not found' }); @@ -71,13 +81,77 @@ export class WebhookController { timestamp: new Date().toISOString(), message: 'Test webhook delivery', }; - await (webhookService as any).deliver(target.endpoint, payload, target.secret); + await getWebhookService().deliver(target.endpoint, payload, target.secret, target.id); return res.status(200).json({ success: true, message: 'Test event delivered' }); } catch (error) { handleError(res, error); } }; + /** + * List delivery attempts for a subscription owned by the caller (Issue #824). + * GET /api/webhooks/:id/deliveries + */ + static listDeliveries = async (req: Request, res: Response) => { + try { + const userId = (req as any).user?.id; + if (!userId) return res.status(401).json({ success: false, message: 'Unauthorized' }); + + const id = req.params.id as string; + const page = Math.max(1, parseInt(req.query.page as string, 10) || 1); + const limit = Math.min(100, Math.max(1, parseInt(req.query.limit as string, 10) || 20)); + + const result = await getWebhookService().listDeliveries(id, userId, page, limit); + return res.status(200).json({ success: true, ...result }); + } catch (error: any) { + if (error.message === 'Webhook subscription not found') { + return res.status(404).json({ success: false, message: error.message }); + } + if (error.message === 'Forbidden: You do not own this webhook subscription') { + return res.status(403).json({ success: false, message: error.message }); + } + handleError(res, error); + } + }; + + /** + * Trigger redelivery of an earlier delivery attempt (Issue #824). + * POST /api/webhooks/:id/deliveries/:deliveryId/redeliver + */ + static redeliver = async (req: Request, res: Response) => { + try { + const userId = (req as any).user?.id; + if (!userId) return res.status(401).json({ success: false, message: 'Unauthorized' }); + + const id = req.params.id as string; + const deliveryId = req.params.deliveryId as string; + + const result = await getWebhookService().redeliver(id, deliveryId, userId); + return res.status(200).json({ + success: true, + message: 'Webhook redelivered successfully', + data: result, + }); + } catch (error: any) { + if ( + error.message === 'Webhook subscription not found' || + error.message === 'Webhook delivery record not found' + ) { + return res.status(404).json({ success: false, message: error.message }); + } + if (error.message === 'Forbidden: You do not own this webhook subscription') { + return res.status(403).json({ success: false, message: error.message }); + } + if (error.message?.includes('No payload available')) { + return res.status(400).json({ success: false, message: error.message }); + } + if (error.message?.includes('Webhook delivery failed after')) { + return res.status(502).json({ success: false, message: error.message }); + } + handleError(res, error); + } + }; + // Privy webhook handlers (Issues #603, #604, #605, #609, #610, #611, #613) static privyUserCreated = async (req: Request, res: Response) => { try { diff --git a/src/entities/WebhookDelivery.ts b/src/entities/WebhookDelivery.ts new file mode 100644 index 00000000..ab30d953 --- /dev/null +++ b/src/entities/WebhookDelivery.ts @@ -0,0 +1,67 @@ +import { + Entity, + PrimaryGeneratedColumn, + Column, + CreateDateColumn, + ManyToOne, + JoinColumn, + Index, +} from 'typeorm'; +import { WebhookSubscription } from './WebhookSubscription'; + +export enum WebhookDeliveryStatus { + PENDING = 'pending', + SUCCEEDED = 'succeeded', + FAILED = 'failed', +} + +/** + * Audit and delivery log for outbound webhook attempts (Issue #824). + * Records response status codes, latency, errors, and stores payload for redelivery. + */ +@Entity('webhook_deliveries') +@Index('IDX_webhook_deliveries_subscriptionId', ['subscriptionId']) +@Index('IDX_webhook_deliveries_createdAt', ['createdAt']) +@Index('IDX_webhook_deliveries_eventId', ['eventId']) +export class WebhookDelivery { + @PrimaryGeneratedColumn('uuid') + id!: string; + + @Column() + subscriptionId!: string; + + @ManyToOne(() => WebhookSubscription, { onDelete: 'CASCADE' }) + @JoinColumn({ name: 'subscriptionId' }) + subscription!: WebhookSubscription; + + @Column() + eventId!: string; + + @Column() + eventType!: string; + + @Column({ type: 'int', default: 1 }) + attempt!: number; + + @Column({ + type: 'varchar', + length: 20, + default: WebhookDeliveryStatus.PENDING, + }) + status!: WebhookDeliveryStatus; + + @Column({ type: 'int', nullable: true }) + responseStatus!: number | null; + + @Column({ type: 'int', nullable: true }) + durationMs!: number | null; + + @Column({ type: 'text', nullable: true }) + error!: string | null; + + @Column({ type: 'simple-json', nullable: true }) + payload!: Record | null; + + @CreateDateColumn() + createdAt!: Date; +} diff --git a/src/entities/WebhookSubscription.ts b/src/entities/WebhookSubscription.ts index 35fa6716..870b546f 100644 --- a/src/entities/WebhookSubscription.ts +++ b/src/entities/WebhookSubscription.ts @@ -40,6 +40,10 @@ export class WebhookSubscription { @Column({ default: true }) isActive!: boolean; + /** Number of consecutive failed event deliveries (auto-disables when threshold exceeded) */ + @Column({ type: 'int', default: 0 }) + consecutiveFailures!: number; + @CreateDateColumn() createdAt!: Date; diff --git a/src/index.ts b/src/index.ts index 30a0c2fa..0c35f723 100644 --- a/src/index.ts +++ b/src/index.ts @@ -8,11 +8,19 @@ import { runSeeders } from './seeders'; import { validateSorobanConfig } from './config/soroban'; import { validateEnvironment } from './config/env'; import { checkAuthFlagConfig, listAuthFlags } from './config/authFlags'; -import { startDbPoolMonitor } from './services/DbPoolMonitor'; -import { startJobQueueWorker, startJobQueueMonitor } from './workers/JobQueueWorker'; +import { startDbPoolMonitor, stopDbPoolMonitor } from './services/DbPoolMonitor'; +import { + startJobQueueWorker, + startJobQueueMonitor, + stopJobQueueMonitor, +} from './workers/JobQueueWorker'; import { registerAiJobHandlers } from './workers/AiJobHandlers'; import logger from './config/logger'; -import { startConnectionStateLogger } from './services/DatabaseConnectionManager'; +import { + startConnectionStateLogger, + stopConnectionStateLogger, +} from './services/DatabaseConnectionManager'; +import { privySessionService } from './services/privy/PrivySessionService'; import { attachProcessHandlers, closeRedis, @@ -83,6 +91,7 @@ async function main() { registerAiJobHandlers(); startJobQueueWorker(); startJobQueueMonitor(); + privySessionService.start(); initRabbitMQ() .then(() => { @@ -98,6 +107,10 @@ async function main() { registerShutdownHook('database', closeDatabase); registerShutdownHook('redis', closeRedis); registerShutdownHook('worker-queues', drainWorkerQueues); + registerShutdownHook('privy-session', () => privySessionService.stop()); + registerShutdownHook('db-pool-monitor', stopDbPoolMonitor); + registerShutdownHook('db-connection-logger', stopConnectionStateLogger); + registerShutdownHook('job-queue-monitor', stopJobQueueMonitor); attachProcessHandlers(); } catch (error) { diff --git a/src/jobs/WebhookRetentionJob.ts b/src/jobs/WebhookRetentionJob.ts new file mode 100644 index 00000000..aee42fa0 --- /dev/null +++ b/src/jobs/WebhookRetentionJob.ts @@ -0,0 +1,34 @@ +import AppDataSource from '../config/db'; +import logger from '../config/logger'; +import { WebhookService } from '../services/WebhookService'; + +/** + * Issue #824: Cleans up webhook delivery records older than 30 days. + * + * Intended to run on a schedule (cron / Kubernetes CronJob) and on demand. + */ +export async function runWebhookRetentionJob(days = 30): Promise { + const service = new WebhookService(); + const deletedCount = await service.cleanupOldDeliveries(days); + logger.info({ deletedCount, days }, 'Webhook delivery retention job finished'); + return deletedCount; +} + +if (require.main === module) { + const daysArg = process.argv.find((arg) => arg.startsWith('--days=')); + const days = daysArg ? Number.parseInt(daysArg.split('=')[1], 10) : 30; + + AppDataSource.initialize() + .then(async () => { + await runWebhookRetentionJob(Number.isFinite(days) ? days : 30); + }) + .finally(async () => { + if (AppDataSource.isInitialized) { + await AppDataSource.destroy(); + } + }) + .catch((error) => { + logger.error({ err: error }, 'Webhook retention job failed'); + process.exit(1); + }); +} diff --git a/src/migrations/1757700000000-AddWebhookDeliveriesAndFailureTracking.ts b/src/migrations/1757700000000-AddWebhookDeliveriesAndFailureTracking.ts new file mode 100644 index 00000000..2e696f05 --- /dev/null +++ b/src/migrations/1757700000000-AddWebhookDeliveriesAndFailureTracking.ts @@ -0,0 +1,144 @@ +import { MigrationInterface, QueryRunner, Table, TableColumn, TableForeignKey, TableIndex } from 'typeorm'; + +/** + * Issue #824: Webhook delivery attempt tracking and failure tracking. + * Creates webhook_deliveries table and adds consecutiveFailures column to webhook_subscriptions. + */ +export class AddWebhookDeliveriesAndFailureTracking1757700000000 implements MigrationInterface { + public async up(queryRunner: QueryRunner): Promise { + const hasSubscriptionsTable = await queryRunner.hasTable('webhook_subscriptions'); + if (hasSubscriptionsTable) { + const subscriptionsTable = await queryRunner.getTable('webhook_subscriptions'); + if (!subscriptionsTable?.findColumnByName('consecutiveFailures')) { + await queryRunner.addColumn( + 'webhook_subscriptions', + new TableColumn({ + name: 'consecutiveFailures', + type: 'int', + default: 0, + }), + ); + } + } + + const hasDeliveriesTable = await queryRunner.hasTable('webhook_deliveries'); + if (!hasDeliveriesTable) { + await queryRunner.createTable( + new Table({ + name: 'webhook_deliveries', + columns: [ + { + name: 'id', + type: 'uuid', + isPrimary: true, + generationStrategy: 'uuid', + default: 'uuid_generate_v4()', + }, + { + name: 'subscriptionId', + type: 'uuid', + }, + { + name: 'eventId', + type: 'varchar', + length: '255', + }, + { + name: 'eventType', + type: 'varchar', + length: '100', + }, + { + name: 'attempt', + type: 'int', + default: 1, + }, + { + name: 'status', + type: 'varchar', + length: '20', + default: "'pending'", + }, + { + name: 'responseStatus', + type: 'int', + isNullable: true, + }, + { + name: 'durationMs', + type: 'int', + isNullable: true, + }, + { + name: 'error', + type: 'text', + isNullable: true, + }, + { + name: 'payload', + type: 'text', + isNullable: true, + }, + { + name: 'createdAt', + type: 'timestamp', + default: 'CURRENT_TIMESTAMP', + }, + ], + }), + true, + ); + + if (hasSubscriptionsTable) { + await queryRunner.createForeignKey( + 'webhook_deliveries', + new TableForeignKey({ + columnNames: ['subscriptionId'], + referencedColumnNames: ['id'], + referencedTableName: 'webhook_subscriptions', + onDelete: 'CASCADE', + }), + ); + } + + await queryRunner.createIndex( + 'webhook_deliveries', + new TableIndex({ + name: 'IDX_webhook_deliveries_subscriptionId', + columnNames: ['subscriptionId'], + }), + ); + + await queryRunner.createIndex( + 'webhook_deliveries', + new TableIndex({ + name: 'IDX_webhook_deliveries_createdAt', + columnNames: ['createdAt'], + }), + ); + + await queryRunner.createIndex( + 'webhook_deliveries', + new TableIndex({ + name: 'IDX_webhook_deliveries_eventId', + columnNames: ['eventId'], + }), + ); + } + } + + public async down(queryRunner: QueryRunner): Promise { + const hasDeliveriesTable = await queryRunner.hasTable('webhook_deliveries'); + if (hasDeliveriesTable) { + await queryRunner.dropTable('webhook_deliveries', true); + } + + const hasSubscriptionsTable = await queryRunner.hasTable('webhook_subscriptions'); + if (hasSubscriptionsTable) { + const subscriptionsTable = await queryRunner.getTable('webhook_subscriptions'); + if (subscriptionsTable?.findColumnByName('consecutiveFailures')) { + await queryRunner.dropColumn('webhook_subscriptions', 'consecutiveFailures'); + } + } + } +} diff --git a/src/routes/SongRoutes.ts b/src/routes/SongRoutes.ts index a29a3396..d4730bc5 100644 --- a/src/routes/SongRoutes.ts +++ b/src/routes/SongRoutes.ts @@ -3,7 +3,7 @@ import { UploadController } from '../controllers/UploadController'; import { validateDTO } from '../middlewares/validate'; import { FinalizeUploadDTO } from '../dtos/FinalizeUploadDTO'; import { UploadChunkDTO } from '../dtos/UploadChunkDTO'; -import { authArtistMiddleware } from '../middlewares/authMiddleware'; +import { authArtistMiddleware, optionalAuth } from '../middlewares/authMiddleware'; import multer from 'multer'; import { CreateCoverDTO } from '../dtos/CreateCoverDTO'; import fs from 'fs'; @@ -130,6 +130,10 @@ router.post('/upload/suggest-tags', authArtistMiddleware, uploadController.sugge router.get('/stream/:id', SongController.streamSong); router.get('/popular', SongController.getPopularSongs); +// Song & Artist statistics (Issue #814) +router.get('/artist/:artistId/stats', optionalAuth, SongController.getArtistStats); +router.get('/:id/stats', optionalAuth, SongController.getSongStats); + // Embeddable player endpoint — public, no auth, same rate limiting as streaming (Redis throttle) router.get('/embed/:id', EmbedController.getSongEmbed); diff --git a/src/routes/artistRoutes.ts b/src/routes/artistRoutes.ts index fee97935..af326396 100644 --- a/src/routes/artistRoutes.ts +++ b/src/routes/artistRoutes.ts @@ -2,8 +2,13 @@ import { Router } from 'express'; import { ArtistProfileController } from '../controllers/ArtistProfileController'; import { ArtistOnChainController } from '../controllers/ArtistOnChainController'; import { ArtistMetadataController } from '../controllers/ArtistMetadataController'; +import { SongController } from '../controllers/SongController'; import { validateDTO } from '../middlewares/validate'; -import { authArtistMiddleware, requireArtistAndVerified } from '../middlewares/authMiddleware'; +import { + authArtistMiddleware, + optionalAuth, + requireArtistAndVerified, +} from '../middlewares/authMiddleware'; import { ConnectStellarWalletDTO } from '../dtos/ConnectStellarWalletDTO'; import { PrepareArtistSetupDTO } from '../dtos/PrepareArtistSetupDTO'; import { SubmitSignedXdrDTO } from '../dtos/SubmitSignedXdrDTO'; @@ -16,6 +21,9 @@ const router = Router(); // Public metadata endpoint for OG tags / JSON-LD (no auth, no private data) router.get('/:id/metadata', ArtistMetadataController.getMetadata); +// Artist statistics (Issue #814) +router.get('/:id/stats', optionalAuth, SongController.getArtistStats); + router.patch( '/update-profile', authArtistMiddleware, diff --git a/src/routes/webhookRoutes.ts b/src/routes/webhookRoutes.ts index 31f87ca2..d7afa72d 100644 --- a/src/routes/webhookRoutes.ts +++ b/src/routes/webhookRoutes.ts @@ -18,6 +18,8 @@ router.post( router.get('/', requireAuth, WebhookController.list); router.delete('/:id', requireAuth, WebhookController.remove); router.post('/:id/test', requireAuth, WebhookController.testDelivery); +router.get('/:id/deliveries', requireAuth, WebhookController.listDeliveries); +router.post('/:id/deliveries/:deliveryId/redeliver', requireAuth, WebhookController.redeliver); // Privy webhook handlers (Issues #603, #604, #605, #606) // These endpoints receive webhooks from Privy for user lifecycle events diff --git a/src/services/AccountDeletionService.ts b/src/services/AccountDeletionService.ts index 6552dfa0..46752f1c 100644 --- a/src/services/AccountDeletionService.ts +++ b/src/services/AccountDeletionService.ts @@ -5,6 +5,7 @@ import logger from '../config/logger'; import { AppError } from '../errors/AppError'; import { CacheService } from './CacheService'; import { User, UserRole } from '../entities/User'; +import { Song } from '../entities/Song'; import { AccountDeletionRequest, type AccountDeletionStatus, @@ -537,10 +538,18 @@ export class AccountDeletionService { // Songs survive erasure, so any cached copy of their streaming manifests is // cleared to drop the former owner's attribution. + // Issue #826: only invalidate that user's songs rather than scanning song:* if (user.role === UserRole.ARTIST) { try { - await CacheService.invalidatePattern('song:*'); - count += 1; + const songRepo = this.dataSource.getRepository(Song); + const userSongs = await songRepo.find({ + where: { artistId: user.id }, + select: ['id'], + }); + for (const song of userSongs) { + await CacheService.clearSong(song.id); + count += 1; + } } catch (error) { warnings.push(`Failed to invalidate song cache: ${(error as Error).message}`); } diff --git a/src/services/CacheService.ts b/src/services/CacheService.ts index d8614462..25072679 100644 --- a/src/services/CacheService.ts +++ b/src/services/CacheService.ts @@ -74,16 +74,36 @@ export class CacheService { } /** - * Remove all keys matching a Redis glob pattern. + * Remove all keys matching a Redis glob pattern using SCAN and UNLINK. * - * @param pattern - Redis key pattern (e.g. "song:*"). + * Replaced KEYS with non-blocking SCAN stream (batch size 500) and UNLINK + * to avoid blocking Redis event loop in production (Issue #826). + * + * @param pattern - Redis key pattern (e.g. "stats:song:123:*"). */ static async invalidatePattern(pattern: string): Promise { try { - const keys = await redis.keys(pattern); - if (keys.length > 0) { - await redis.del(...keys); - logger.info({ pattern, keyCount: keys.length }, 'Cache pattern invalidated'); + const stream = redis.scanStream({ + match: pattern, + count: 500, + }); + + let totalKeys = 0; + + for await (const resultKeys of stream) { + const keys = resultKeys as string[]; + if (keys && keys.length > 0) { + totalKeys += keys.length; + if (typeof redis.unlink === 'function') { + await redis.unlink(...keys); + } else { + await redis.del(...keys); + } + } + } + + if (totalKeys > 0) { + logger.info({ pattern, keyCount: totalKeys }, 'Cache pattern invalidated'); } } catch (err) { logger.warn({ err, pattern }, 'Cache pattern invalidate failed'); diff --git a/src/services/DatabaseConnectionManager.ts b/src/services/DatabaseConnectionManager.ts index 496a08c1..a4bf392a 100644 --- a/src/services/DatabaseConnectionManager.ts +++ b/src/services/DatabaseConnectionManager.ts @@ -96,7 +96,9 @@ export async function attemptReconnect( return false; } -export function createDbCircuitBreaker(dataSource: DataSource): CircuitBreaker { +let connectionStateLoggerTimer: NodeJS.Timeout | null = null; + +export function createDbCircuitBreaker(dataSource: DataSource, autoStart = false): CircuitBreaker { const cb = new CircuitBreaker( async () => { if (!dataSource.isInitialized) return false; @@ -109,14 +111,18 @@ export function createDbCircuitBreaker(dataSource: DataSource): CircuitBreaker { }, ); - cb.startHealthChecks(); + if (autoStart) { + cb.startHealthChecks(); + } return cb; } export function startConnectionStateLogger(): NodeJS.Timeout { + if (connectionStateLoggerTimer) return connectionStateLoggerTimer; + const intervalMs = Number(process.env.DB_CONNECTION_STATE_LOG_INTERVAL_MS || 60000); - const timer = setInterval(() => { + connectionStateLoggerTimer = setInterval(() => { logger.info( { connected: dbConnectionState.connected, @@ -129,9 +135,17 @@ export function startConnectionStateLogger(): NodeJS.Timeout { ); }, intervalMs); - if (timer.unref) { - timer.unref(); + if (typeof connectionStateLoggerTimer.unref === 'function') { + connectionStateLoggerTimer.unref(); } - return timer; + return connectionStateLoggerTimer; } + +export function stopConnectionStateLogger(): void { + if (connectionStateLoggerTimer) { + clearInterval(connectionStateLoggerTimer); + connectionStateLoggerTimer = null; + } +} + diff --git a/src/services/WebhookService.ts b/src/services/WebhookService.ts index 5423ae8b..3cd358b8 100644 --- a/src/services/WebhookService.ts +++ b/src/services/WebhookService.ts @@ -2,6 +2,8 @@ import crypto from 'crypto'; import { Repository } from 'typeorm'; import AppDataSource from '../config/db'; import { WebhookSubscription } from '../entities/WebhookSubscription'; +import { WebhookDelivery, WebhookDeliveryStatus } from '../entities/WebhookDelivery'; +import { NotificationService } from './NotificationService'; import logger from '../config/logger'; export interface WebhookPayload { @@ -16,23 +18,69 @@ function sleep(ms: number): Promise { } /** - * Service for managing webhook subscriptions and delivering signed event payloads. - * Follows WEBHOOK_IMPLEMENTATION_PLAN.md: HMAC-SHA256 signing + exponential backoff retries. + * Service for managing webhook subscriptions, recording delivery attempts, + * and dispatching signed event payloads with retry backoff and failure isolation (Issue #824). */ export class WebhookService { private subscriptionRepo: Repository; + private deliveryRepo: Repository; + private notificationService: NotificationService; /** Allows injection of fetch for testing; defaults to global fetch */ private fetchFn: typeof fetch; - constructor(fetchFn?: typeof fetch) { - try { - this.subscriptionRepo = AppDataSource.getRepository(WebhookSubscription); - } catch { - // In tests AppDataSource may be mocked without this entity — create a dummy stub - this.subscriptionRepo = { - find: async () => [], - } as unknown as Repository; + constructor( + fetchFn?: typeof fetch, + subscriptionRepo?: Repository, + deliveryRepo?: Repository, + notificationService?: NotificationService, + ) { + if (subscriptionRepo) { + this.subscriptionRepo = subscriptionRepo; + } else { + try { + this.subscriptionRepo = AppDataSource.getRepository(WebhookSubscription); + } catch { + this.subscriptionRepo = { + find: async () => [], + } as unknown as Repository; + } + } + + if (deliveryRepo) { + this.deliveryRepo = deliveryRepo; + } else { + try { + this.deliveryRepo = AppDataSource.getRepository(WebhookDelivery); + } catch { + this.deliveryRepo = { + save: async (entity: any) => entity, + findAndCount: async () => [[], 0], + findOne: async () => null, + createQueryBuilder: () => ({ + delete: () => ({ + from: () => ({ + where: () => ({ + execute: async () => ({ affected: 0 }), + }), + }), + }), + }), + } as unknown as Repository; + } } + + if (notificationService) { + this.notificationService = notificationService; + } else { + try { + this.notificationService = new NotificationService(); + } catch { + this.notificationService = { + create: async () => ({} as any), + } as unknown as NotificationService; + } + } + this.fetchFn = fetchFn || (global.fetch as typeof fetch); } @@ -78,6 +126,7 @@ export class WebhookService { secret, eventTypes: eventTypes && eventTypes.length > 0 ? eventTypes : ['*'], isActive: true, + consecutiveFailures: 0, }); const saved = await this.subscriptionRepo.save(sub); @@ -97,15 +146,39 @@ export class WebhookService { /** * Deliver payload to a single endpoint with retries and exponential backoff. - * Mirrors plan's WebhookService.deliver implementation. + * Records every attempt into WebhookDelivery and handles auto-disable tracking (Issue #824). */ - async deliver(endpoint: string, payload: WebhookPayload, secret: string): Promise { + async deliver( + endpoint: string, + payload: WebhookPayload, + secret: string, + subscriptionId?: string, + ): Promise { const maxRetries = 3; const signature = this.signPayload(payload, secret); + let sub: WebhookSubscription | null = null; + let subId = subscriptionId; + if (subId && this.subscriptionRepo?.findOne) { + try { + sub = await this.subscriptionRepo.findOne({ where: { id: subId } }); + } catch {} + } else if (!subId && this.subscriptionRepo?.findOne) { + try { + sub = await this.subscriptionRepo.findOne({ where: { endpoint } }); + if (sub) subId = sub.id; + } catch {} + } + + let lastDelivery: WebhookDelivery | undefined; + for (let attempt = 0; attempt < maxRetries; attempt++) { + const startTime = Date.now(); + let res: Response | null = null; + let fetchError: Error | null = null; + try { - const res = await this.fetchFn(endpoint, { + res = await this.fetchFn(endpoint, { method: 'POST', headers: { 'Content-Type': 'application/json', @@ -113,34 +186,117 @@ export class WebhookService { }, body: JSON.stringify(payload), }); - - if (res.ok) { - logger.info( - { endpoint, eventType: payload.eventType, attempt: attempt + 1 }, - 'Webhook delivered', - ); - return; - } - logger.warn( - { endpoint, status: res.status, attempt: attempt + 1 }, - 'Webhook delivery failed - non-OK status', - ); } catch (error) { - logger.warn( - { endpoint, attempt: attempt + 1, error: (error as Error).message }, - 'Webhook delivery attempt failed', + fetchError = error as Error; + } + + const durationMs = Date.now() - startTime; + const attemptNum = attempt + 1; + + if (res && res.ok) { + logger.info( + { endpoint, eventType: payload.eventType, attempt: attemptNum }, + 'Webhook delivered', ); + + // Record successful delivery attempt + if (subId && this.deliveryRepo?.save) { + try { + lastDelivery = await this.deliveryRepo.save({ + subscriptionId: subId, + eventId: payload.eventId || crypto.randomUUID(), + eventType: payload.eventType || 'unknown', + attempt: attemptNum, + status: WebhookDeliveryStatus.SUCCEEDED, + responseStatus: res.status, + durationMs, + error: null, + payload, + }); + } catch (err) { + logger.warn({ err }, 'Failed to record webhook delivery row'); + } + } + + // Reset consecutive failures on success + if (sub && sub.consecutiveFailures > 0 && this.subscriptionRepo?.save) { + try { + sub.consecutiveFailures = 0; + await this.subscriptionRepo.save(sub); + } catch (err) { + logger.warn({ err }, 'Failed to reset webhook consecutive failures'); + } + } + + return lastDelivery; + } + + const responseStatus = res ? res.status : null; + const errorMsg = fetchError ? fetchError.message : `HTTP ${res?.status}: ${res?.statusText}`; + + logger.warn( + { endpoint, status: responseStatus, attempt: attemptNum, error: errorMsg }, + 'Webhook delivery attempt failed', + ); + + // Record failed delivery attempt + if (subId && this.deliveryRepo?.save) { + try { + lastDelivery = await this.deliveryRepo.save({ + subscriptionId: subId, + eventId: payload.eventId || crypto.randomUUID(), + eventType: payload.eventType || 'unknown', + attempt: attemptNum, + status: WebhookDeliveryStatus.FAILED, + responseStatus, + durationMs, + error: errorMsg, + payload, + }); + } catch (err) { + logger.warn({ err }, 'Failed to record webhook delivery row'); + } } if (attempt < maxRetries - 1) { // Use shorter backoff in test env to keep tests fast const baseMs = process.env.NODE_ENV === 'test' ? 10 : 1000; - const backoff = 2 ** attempt * baseMs; // Exponential backoff: 1s, 2s, 4s (10ms in test) + const backoff = 2 ** attempt * baseMs; await sleep(backoff); } } - // After exhausting retries, log as dead-letter + // After exhausting all retries, increment consecutive failures & check auto-disable threshold + if (sub && this.subscriptionRepo?.save) { + try { + sub.consecutiveFailures = (sub.consecutiveFailures || 0) + 1; + const maxFailures = Number(process.env.WEBHOOK_MAX_CONSECUTIVE_FAILURES) || 5; + if (sub.consecutiveFailures >= maxFailures && sub.isActive) { + sub.isActive = false; + logger.warn( + { subscriptionId: sub.id, endpoint: sub.endpoint, failures: sub.consecutiveFailures }, + 'Webhook subscription disabled due to consecutive failures', + ); + if (this.notificationService?.create) { + await this.notificationService.create({ + userId: sub.userId, + type: 'system', + title: 'Webhook subscription disabled', + message: `Your webhook subscription to ${sub.endpoint} has been disabled after ${sub.consecutiveFailures} consecutive failed events.`, + data: { + subscriptionId: sub.id, + endpoint: sub.endpoint, + consecutiveFailures: sub.consecutiveFailures, + }, + }); + } + } + await this.subscriptionRepo.save(sub); + } catch (err) { + logger.warn({ err }, 'Failed to update subscription consecutive failures'); + } + } + logger.error( { endpoint, eventType: payload.eventType }, 'Webhook delivery exhausted retries — dead letter', @@ -148,6 +304,96 @@ export class WebhookService { throw new Error(`Webhook delivery failed after ${maxRetries} attempts to ${endpoint}`); } + /** + * List deliveries for a subscription owned by the requesting user, newest first. + * Time Complexity: O(limit) with indexed (subscriptionId, createdAt). + * Space Complexity: O(limit) in memory. + */ + async listDeliveries( + subscriptionId: string, + userId: string, + page = 1, + limit = 20, + ): Promise<{ + data: WebhookDelivery[]; + pagination: { page: number; limit: number; total: number; totalPages: number }; + }> { + const sub = await this.subscriptionRepo.findOne({ where: { id: subscriptionId } }); + if (!sub) { + throw new Error('Webhook subscription not found'); + } + if (sub.userId !== userId) { + throw new Error('Forbidden: You do not own this webhook subscription'); + } + + const safePage = Math.max(1, page); + const safeLimit = Math.min(Math.max(1, limit), 100); + + const [items, total] = await this.deliveryRepo.findAndCount({ + where: { subscriptionId }, + order: { createdAt: 'DESC' }, + skip: (safePage - 1) * safeLimit, + take: safeLimit, + }); + + return { + data: items, + pagination: { + page: safePage, + limit: safeLimit, + total, + totalPages: Math.ceil(total / safeLimit), + }, + }; + } + + /** + * Trigger redelivery of a previous delivery payload for an owned subscription. + */ + async redeliver( + subscriptionId: string, + deliveryId: string, + userId: string, + ): Promise { + const sub = await this.subscriptionRepo.findOne({ where: { id: subscriptionId } }); + if (!sub) { + throw new Error('Webhook subscription not found'); + } + if (sub.userId !== userId) { + throw new Error('Forbidden: You do not own this webhook subscription'); + } + + const delivery = await this.deliveryRepo.findOne({ + where: { id: deliveryId, subscriptionId }, + }); + if (!delivery) { + throw new Error('Webhook delivery record not found'); + } + + if (!delivery.payload) { + throw new Error('No payload available for this webhook delivery'); + } + + // Trigger redelivery using original payload and credentials + return await this.deliver(sub.endpoint, delivery.payload as WebhookPayload, sub.secret, sub.id); + } + + /** + * Delete delivery logs older than the retention period (default 30 days). + * Time Complexity: O(k) indexed by createdAt. + * Space Complexity: O(1). + */ + async cleanupOldDeliveries(retentionDays = 30): Promise { + const cutoff = new Date(Date.now() - retentionDays * 24 * 60 * 60 * 1000); + const result = await this.deliveryRepo + .createQueryBuilder() + .delete() + .from(WebhookDelivery) + .where('createdAt < :cutoff', { cutoff }) + .execute(); + return result.affected || 0; + } + /** * Publish an event to all matching subscriptions. * Called at key state transitions: song minted, sale completed. @@ -167,7 +413,6 @@ export class WebhookService { } // Fetch active subscriptions that match this event - // For simplicity, load all active and filter in-memory (simple-array matching) let allActive: WebhookSubscription[] = []; try { allActive = await this.subscriptionRepo.find({ where: { isActive: true } }); @@ -191,7 +436,7 @@ export class WebhookService { // Deliver to each matched subscription (fire-and-forget with logging; don't fail caller on delivery failure) await Promise.allSettled( matched.map((sub) => - this.deliver(sub.endpoint, payload, sub.secret).catch((err) => { + this.deliver(sub.endpoint, payload, sub.secret, sub.id).catch((err) => { logger.error( { endpoint: sub.endpoint, eventType, err: err.message }, 'Webhook publish delivery failed', @@ -204,7 +449,6 @@ export class WebhookService { /** Backwards-compatible alias for plan's event name */ async emitMintStatusChanged(songId: string, details: Record): Promise { await this.publish('song.minted', { songId, ...details }); - // Also emit legacy event name for subscribers using old convention await this.publish('mint_status_changed', { songId, ...details }); } @@ -215,3 +459,4 @@ export class WebhookService { } export default WebhookService; + diff --git a/src/services/__tests__/AccountDeletionService.test.ts b/src/services/__tests__/AccountDeletionService.test.ts index f4f6fb70..20da26d7 100644 --- a/src/services/__tests__/AccountDeletionService.test.ts +++ b/src/services/__tests__/AccountDeletionService.test.ts @@ -5,7 +5,7 @@ jest.mock('../../config/db', () => ({ default: { getRepository: jest.fn() }, })); jest.mock('../CacheService', () => ({ - CacheService: { invalidate: jest.fn(), invalidatePattern: jest.fn() }, + CacheService: { invalidate: jest.fn(), invalidatePattern: jest.fn(), clearSong: jest.fn() }, })); jest.mock('../privy/PrivyIdentityService', () => ({ deleteLinkedPrivyIdentity: jest.fn(), @@ -14,6 +14,7 @@ jest.mock('../privy/PrivyIdentityService', () => ({ import { DataSource } from 'typeorm'; import { AccountDeletionService } from '../AccountDeletionService'; import { User, UserRole } from '../../entities/User'; +import { Song } from '../../entities/Song'; import { CacheService } from '../CacheService'; import { deleteLinkedPrivyIdentity } from '../privy/PrivyIdentityService'; @@ -63,6 +64,7 @@ describe('AccountDeletionService (#633)', () => { /** Every SQL statement issued, for assertions. */ let statements: Array<{ sql: string; params?: unknown[] }>; let refreshTokenRepo: { update: jest.Mock }; + let songRepo: { find: jest.Mock }; beforeEach(() => { jest.clearAllMocks(); @@ -82,6 +84,7 @@ describe('AccountDeletionService (#633)', () => { }; refreshTokenRepo = { update: jest.fn().mockResolvedValue({ affected: 1 }) }; + songRepo = { find: jest.fn().mockResolvedValue([{ id: 'song-1' }, { id: 'song-2' }]) }; manager = { query: jest.fn((sql: string, params?: unknown[]) => { @@ -91,6 +94,7 @@ describe('AccountDeletionService (#633)', () => { // Mirrors the repositories the service touches inside its transaction. getRepository: jest.fn((entity: unknown) => { if (entity === User) return userRepo; + if (entity === Song) return songRepo; if (entity === 'refresh_tokens') return refreshTokenRepo; return requestRepo; }), @@ -100,6 +104,7 @@ describe('AccountDeletionService (#633)', () => { dataSource = { getRepository: jest.fn((entity: unknown) => { if (entity === User) return userRepo; + if (entity === Song) return songRepo; return requestRepo; }), query: jest.fn((sql: string, params?: unknown[]) => { @@ -426,7 +431,7 @@ describe('AccountDeletionService (#633)', () => { expect(result.privyDeletion).toBe('skipped'); }); - it('invalidates the artist cache so cached copies lose the attribution', async () => { + it('invalidates the artist cache and only that artist’s songs so cached copies lose attribution', async () => { userRepo.findOneBy.mockResolvedValue( makeUser({ role: UserRole.ARTIST, @@ -438,7 +443,13 @@ describe('AccountDeletionService (#633)', () => { await service.purgeUser('user-1'); expect(CacheService.invalidate).toHaveBeenCalledWith('artist:user-1'); - expect(CacheService.invalidatePattern).toHaveBeenCalledWith('song:*'); + expect(songRepo.find).toHaveBeenCalledWith({ + where: { artistId: 'user-1' }, + select: ['id'], + }); + expect(CacheService.clearSong).toHaveBeenCalledWith('song-1'); + expect(CacheService.clearSong).toHaveBeenCalledWith('song-2'); + expect(CacheService.invalidatePattern).not.toHaveBeenCalledWith('song:*'); }); }); diff --git a/src/services/privy/PrivySessionService.ts b/src/services/privy/PrivySessionService.ts index e7849931..78fa7254 100644 --- a/src/services/privy/PrivySessionService.ts +++ b/src/services/privy/PrivySessionService.ts @@ -29,8 +29,26 @@ class PrivySessionService { private cleanupInterval: NodeJS.Timeout | null = null; constructor() { - // Clean up expired revocations every 5 minutes + // Timers are started via explicit start() to prevent leaking timers in tests (Issue #834) + } + + /** + * Start periodic cleanup of expired revocations. + * Calls .unref() so the timer does not keep Node.js or Jest workers alive. + */ + start(): void { + if (this.cleanupInterval) return; this.cleanupInterval = setInterval(() => this.cleanup(), 5 * 60 * 1000); + if (typeof this.cleanupInterval.unref === 'function') { + this.cleanupInterval.unref(); + } + } + + /** + * Stop the cleanup interval. Alias for shutdown(). + */ + stop(): void { + this.shutdown(); } /** diff --git a/tests/setup/jestTeardown.ts b/tests/setup/jestTeardown.ts new file mode 100644 index 00000000..5d8258f9 --- /dev/null +++ b/tests/setup/jestTeardown.ts @@ -0,0 +1,46 @@ +/** + * Global Jest teardown hook (Issue #834). + * Ensures in-memory timers and shared Redis/DB handles are cleanly terminated + * after each test file completes so worker processes exit immediately without + * leaking open handles. + */ +afterAll(async () => { + try { + const privyPath = require.resolve('../../src/services/privy/PrivySessionService'); + if (require.cache[privyPath]) { + require('../../src/services/privy/PrivySessionService').privySessionService?.stop?.(); + } + } catch {} + + try { + const poolPath = require.resolve('../../src/services/DbPoolMonitor'); + if (require.cache[poolPath]) { + require('../../src/services/DbPoolMonitor').stopDbPoolMonitor?.(); + } + } catch {} + + try { + const dbConnPath = require.resolve('../../src/services/DatabaseConnectionManager'); + if (require.cache[dbConnPath]) { + require('../../src/services/DatabaseConnectionManager').stopConnectionStateLogger?.(); + } + } catch {} + + try { + const workerPath = require.resolve('../../src/workers/JobQueueWorker'); + if (require.cache[workerPath]) { + require('../../src/workers/JobQueueWorker').stopJobQueueMonitor?.(); + } + } catch {} + + try { + const redisPath = require.resolve('../../src/config/redis'); + if (require.cache[redisPath]) { + const redisModule = require('../../src/config/redis'); + const redisInstance = redisModule.default || redisModule; + if (redisInstance && typeof redisInstance.disconnect === 'function') { + redisInstance.disconnect(false); + } + } + } catch {} +});