diff --git a/.env.example b/.env.example index a1ac7a0..b238e39 100644 --- a/.env.example +++ b/.env.example @@ -25,6 +25,12 @@ CONFIRMATIONS=12 OPENSEA_API_KEY= OPENSEA_STREAM_URL=wss://stream.openseabeta.com/socket +# Web Push notifications (optional; generate VAPID keys with `npx web-push generate-vapid-keys`) +WEB_PUSH_VAPID_PUBLIC_KEY= +WEB_PUSH_VAPID_PRIVATE_KEY= +WEB_PUSH_SUBJECT=mailto:noreply@grails.market +WEB_PUSH_TTL_SECONDS=86400 + # API API_PORT=3000 API_HOST=0.0.0.0 @@ -55,4 +61,4 @@ WEB2_TLD_DATA_API_KEY= # valuation_settings DB row (cat-admin → Valuation settings), not an env var. # Node Environment -NODE_ENV=development \ No newline at end of file +NODE_ENV=development diff --git a/.prettierrc b/.prettierrc new file mode 100644 index 0000000..c2b7a57 --- /dev/null +++ b/.prettierrc @@ -0,0 +1,8 @@ +{ + "semi": true, + "singleQuote": true, + "trailingComma": "all", + "printWidth": 120, + "tabWidth": 2, + "plugins": [] +} diff --git a/package-lock.json b/package-lock.json index 6812f3d..4a621f0 100644 --- a/package-lock.json +++ b/package-lock.json @@ -5379,6 +5379,16 @@ "resolved": "https://registry.npmjs.org/@types/unist/-/unist-3.0.3.tgz", "integrity": "sha512-ko/gIFJRv177XgZsZcBwnqJN5x/Gien8qNOn0D5bQU/zAzVf9Zt3BlcUiLqhV9y4ARk0GbT3tnUiPNgnTXzc/Q==" }, + "node_modules/@types/web-push": { + "version": "3.6.4", + "resolved": "https://registry.npmjs.org/@types/web-push/-/web-push-3.6.4.tgz", + "integrity": "sha512-GnJmSr40H3RAnj0s34FNTcJi1hmWFV5KXugE0mYWnYhgTAHLJ/dJKAwDmvPJYMke0RplY2XE9LnM4hqSqKIjhQ==", + "dev": true, + "license": "MIT", + "dependencies": { + "@types/node": "*" + } + }, "node_modules/@types/ws": { "version": "8.18.1", "dev": true, @@ -5740,6 +5750,15 @@ "version": "4.0.0-beta.5", "license": "MIT" }, + "node_modules/agent-base": { + "version": "7.1.4", + "resolved": "https://registry.npmjs.org/agent-base/-/agent-base-7.1.4.tgz", + "integrity": "sha512-MnA+YT8fwfJPgBx3m60MNqakm30XOkyIoH1y6huTQvC0PwZG7ki8NacLBcrPbNoo8vEZy7Jpuk7+jMO+CUovTQ==", + "license": "MIT", + "engines": { + "node": ">= 14" + } + }, "node_modules/aggregate-error": { "version": "3.1.0", "license": "MIT", @@ -5955,6 +5974,18 @@ "node": ">=8" } }, + "node_modules/asn1.js": { + "version": "5.4.1", + "resolved": "https://registry.npmjs.org/asn1.js/-/asn1.js-5.4.1.tgz", + "integrity": "sha512-+I//4cYPccV8LdmBLiX8CYvf9Sp3vQsrqu2QNXRcrbiWvcx/UdlFiqUJJzxRQxgsZmvhXhn4cSKeSmoFjVdupA==", + "license": "MIT", + "dependencies": { + "bn.js": "^4.0.0", + "inherits": "^2.0.1", + "minimalistic-assert": "^1.0.0", + "safer-buffer": "^2.1.0" + } + }, "node_modules/assertion-error": { "version": "1.1.0", "resolved": "https://registry.npmjs.org/assertion-error/-/assertion-error-1.1.0.tgz", @@ -6405,6 +6436,12 @@ "url": "https://github.com/sponsors/sindresorhus" } }, + "node_modules/bn.js": { + "version": "4.12.3", + "resolved": "https://registry.npmjs.org/bn.js/-/bn.js-4.12.3.tgz", + "integrity": "sha512-fGTi3gxV/23FTYdAoUtLYp6qySe2KE3teyZitipKNRuVYcBkoP/bB3guXN/XVKUe9mxCHXnc9C4ocyz8OmgN0g==", + "license": "MIT" + }, "node_modules/bowser": { "version": "2.12.1", "license": "MIT" @@ -9383,6 +9420,15 @@ "url": "https://github.com/fb55/entities?sponsor=1" } }, + "node_modules/http_ece": { + "version": "1.2.0", + "resolved": "https://registry.npmjs.org/http_ece/-/http_ece-1.2.0.tgz", + "integrity": "sha512-JrF8SSLVmcvc5NducxgyOrKXe3EsyHMgBFgSaIUGmArKe+rwr0uphRkRXvwiom3I+fpIfoItveHrfudL8/rxuA==", + "license": "MIT", + "engines": { + "node": ">=16" + } + }, "node_modules/http-cache-semantics": { "version": "4.2.0", "resolved": "https://registry.npmjs.org/http-cache-semantics/-/http-cache-semantics-4.2.0.tgz", @@ -9402,6 +9448,19 @@ "node": ">= 0.8" } }, + "node_modules/https-proxy-agent": { + "version": "7.0.6", + "resolved": "https://registry.npmjs.org/https-proxy-agent/-/https-proxy-agent-7.0.6.tgz", + "integrity": "sha512-vK9P5/iUfdl95AI+JVyUuIcVtd4ofvtrOr3HNtM2yxC9bnMbEdp3x01OhQNnjb8IJYi38VlTE3mBXwcfvywuSw==", + "license": "MIT", + "dependencies": { + "agent-base": "^7.1.2", + "debug": "4" + }, + "engines": { + "node": ">= 14" + } + }, "node_modules/human-signals": { "version": "2.1.0", "resolved": "https://registry.npmjs.org/human-signals/-/human-signals-2.1.0.tgz", @@ -11332,6 +11391,12 @@ "url": "https://github.com/sponsors/sindresorhus" } }, + "node_modules/minimalistic-assert": { + "version": "1.0.1", + "resolved": "https://registry.npmjs.org/minimalistic-assert/-/minimalistic-assert-1.0.1.tgz", + "integrity": "sha512-UtJcAD4yEaGtjPezWuO9wC4nwUnVH/8/Im3yEHQP4b67cXlD/Qr9hdITCU1xDbSEXg2XKNaP8jsReV7vQd00/A==", + "license": "ISC" + }, "node_modules/minimatch": { "version": "9.0.3", "dev": true, @@ -13443,6 +13508,12 @@ "node": ">=10" } }, + "node_modules/safer-buffer": { + "version": "2.1.2", + "resolved": "https://registry.npmjs.org/safer-buffer/-/safer-buffer-2.1.2.tgz", + "integrity": "sha512-YZo3K82SD7Riyi0E1EQPojLz7kpepnSQI9IyPbHHg1XXXevb5dJI7tpyN2ADxGcQbHG7vcyRHk0cbwqcQriUtg==", + "license": "MIT" + }, "node_modules/sanitize-html": { "version": "2.17.3", "resolved": "https://registry.npmjs.org/sanitize-html/-/sanitize-html-2.17.3.tgz", @@ -15463,6 +15534,46 @@ "url": "https://github.com/sponsors/wooorm" } }, + "node_modules/web-push": { + "version": "3.6.7", + "resolved": "https://registry.npmjs.org/web-push/-/web-push-3.6.7.tgz", + "integrity": "sha512-OpiIUe8cuGjrj3mMBFWY+e4MMIkW3SVT+7vEIjvD9kejGUypv8GPDf84JdPWskK8zMRIJ6xYGm+Kxr8YkPyA0A==", + "license": "MPL-2.0", + "dependencies": { + "asn1.js": "^5.3.0", + "http_ece": "1.2.0", + "https-proxy-agent": "^7.0.0", + "jws": "^4.0.0", + "minimist": "^1.2.5" + }, + "bin": { + "web-push": "src/cli.js" + }, + "engines": { + "node": ">= 16" + } + }, + "node_modules/web-push/node_modules/jwa": { + "version": "2.0.1", + "resolved": "https://registry.npmjs.org/jwa/-/jwa-2.0.1.tgz", + "integrity": "sha512-hRF04fqJIP8Abbkq5NKGN0Bbr3JxlQ+qhZufXVr0DvujKy93ZCbXZMHDL4EOtodSbCWxOqR8MS1tXA5hwqCXDg==", + "license": "MIT", + "dependencies": { + "buffer-equal-constant-time": "^1.0.1", + "ecdsa-sig-formatter": "1.0.11", + "safe-buffer": "^5.0.1" + } + }, + "node_modules/web-push/node_modules/jws": { + "version": "4.0.1", + "resolved": "https://registry.npmjs.org/jws/-/jws-4.0.1.tgz", + "integrity": "sha512-EKI/M/yqPncGUUh44xz0PxSidXFr/+r0pA70+gIYhjv+et7yxM+s29Y+VGDkovRofQem0fs7Uvf4+YmAdyRduA==", + "license": "MIT", + "dependencies": { + "jwa": "^2.0.1", + "safe-buffer": "^5.0.1" + } + }, "node_modules/which": { "version": "2.0.2", "license": "ISC", @@ -16043,11 +16154,13 @@ "nodemailer": "^7.0.9", "pg-boss": "^9.0.3", "pino": "^8.17.2", - "pino-pretty": "^10.3.1" + "pino-pretty": "^10.3.1", + "web-push": "^3.6.7" }, "devDependencies": { "@types/node": "^20.10.0", "@types/nodemailer": "^7.0.2", + "@types/web-push": "^3.6.4", "tsx": "^4.7.0", "typescript": "^5.3.0" } diff --git a/services/api/migrations/seq/0894_create_push_subscriptions.sql b/services/api/migrations/seq/0894_create_push_subscriptions.sql new file mode 100644 index 0000000..bb9cac5 --- /dev/null +++ b/services/api/migrations/seq/0894_create_push_subscriptions.sql @@ -0,0 +1,42 @@ +-- Migration: Create push subscriptions +-- Date: 2026-06-21 +-- Description: Stores browser Web Push subscriptions for authenticated users + +CREATE TABLE IF NOT EXISTS push_subscriptions ( + id SERIAL PRIMARY KEY, + user_id INTEGER NOT NULL REFERENCES users(id) ON DELETE CASCADE, + endpoint TEXT NOT NULL UNIQUE, + p256dh TEXT NOT NULL, + auth TEXT NOT NULL, + expiration_time TIMESTAMP, + device_name VARCHAR(120), + user_agent TEXT, + enabled BOOLEAN NOT NULL DEFAULT TRUE, + last_seen_at TIMESTAMP NOT NULL DEFAULT NOW(), + created_at TIMESTAMP NOT NULL DEFAULT NOW(), + updated_at TIMESTAMP NOT NULL DEFAULT NOW() +); + +CREATE INDEX IF NOT EXISTS idx_push_subscriptions_user +ON push_subscriptions(user_id); + +CREATE INDEX IF NOT EXISTS idx_push_subscriptions_enabled_user +ON push_subscriptions(user_id) +WHERE enabled = TRUE; + +CREATE INDEX IF NOT EXISTS idx_push_subscriptions_endpoint +ON push_subscriptions(endpoint); + +CREATE OR REPLACE FUNCTION update_push_subscriptions_updated_at() +RETURNS TRIGGER AS $$ +BEGIN + NEW.updated_at = NOW(); + RETURN NEW; +END; +$$ LANGUAGE plpgsql; + +DROP TRIGGER IF EXISTS update_push_subscriptions_updated_at ON push_subscriptions; +CREATE TRIGGER update_push_subscriptions_updated_at + BEFORE UPDATE ON push_subscriptions + FOR EACH ROW + EXECUTE FUNCTION update_push_subscriptions_updated_at(); diff --git a/services/api/src/routes/index.ts b/services/api/src/routes/index.ts index 9122adb..25751f5 100644 --- a/services/api/src/routes/index.ts +++ b/services/api/src/routes/index.ts @@ -43,11 +43,14 @@ import { blocksRoutes } from './blocks'; import { commentsRoutes } from './comments'; import { feedRoutes } from './feed'; import { valuationsRoutes } from './valuations'; +import { pushRoutes, userPushSubscriptionRoutes } from './push'; export function registerRoutes(fastify: FastifyInstance) { fastify.register(healthRoutes, { prefix: '/health' }); fastify.register(authRoutes, { prefix: '/api/v1/auth' }); + fastify.register(userPushSubscriptionRoutes, { prefix: '/api/v1/users/me/push-subscriptions' }); fastify.register(usersRoutes, { prefix: '/api/v1/users' }); + fastify.register(pushRoutes, { prefix: '/api/v1/push' }); fastify.register(verificationRoutes, { prefix: '/api/v1/verification' }); fastify.register(watchlistRoutes, { prefix: '/api/v1/watchlist' }); fastify.register(notificationsRoutes, { prefix: '/api/v1/notifications' }); @@ -90,4 +93,4 @@ export function registerRoutes(fastify: FastifyInstance) { fastify.register(commentsRoutes, { prefix: '/api/v1/comments' }); fastify.register(valuationsRoutes, { prefix: '/api/v1/valuations' }); fastify.register(websocketRoutes, { prefix: '/ws' }); -} \ No newline at end of file +} diff --git a/services/api/src/routes/push.ts b/services/api/src/routes/push.ts new file mode 100644 index 0000000..1550549 --- /dev/null +++ b/services/api/src/routes/push.ts @@ -0,0 +1,459 @@ +import type { FastifyInstance } from 'fastify' +import dns from 'dns/promises' +import net from 'net' +import { z } from 'zod' +import { config, getPostgresPool, type APIResponse } from '../../../shared/src' +import { requireAuth } from '../middleware/auth' + +const MAX_PUSH_SUBSCRIPTIONS_PER_USER = 20 // Change this as you wish + +const PushSubscriptionBodySchema = z.object({ + endpoint: z.string().url().max(2048), + keys: z.object({ + p256dh: z.string().min(43).max(256), + auth: z.string().min(16).max(256), + }), + expirationTime: z.union([z.number(), z.string(), z.null()]).optional(), + deviceName: z.string().trim().min(1).max(120).optional(), +}) + +const SubscriptionParamsSchema = z.object({ + id: z.coerce.number().int().positive(), +}) + +interface PushSubscriptionRow { + id: number + endpoint: string + device_name: string | null + enabled: boolean + last_seen_at: Date + created_at: Date +} + +function serializeSubscription(row: PushSubscriptionRow) { + return { + id: row.id, + endpoint: row.endpoint, + deviceName: row.device_name, + enabled: row.enabled, + lastSeenAt: row.last_seen_at, + createdAt: row.created_at, + } +} + +function parseExpirationTime( + value: string | number | null | undefined, +): Date | null { + if (value === undefined || value === null) { + return null + } + + const timestamp = typeof value === 'string' ? Number(value) : value + if (!Number.isFinite(timestamp) || timestamp <= 0) { + return null + } + + return new Date(timestamp) +} + +function getUserAgent(header: string | string[] | undefined): string | null { + if (Array.isArray(header)) { + return header.join(' ') + } + + return header || null +} + +function isBlockedIpAddress(address: string): boolean { + if (net.isIPv4(address)) { + const parts = address.split('.').map(Number) + const [first, second] = parts + + return ( + first === 0 || + first === 10 || + first === 127 || + (first === 169 && second === 254) || + (first === 172 && second >= 16 && second <= 31) || + (first === 192 && second === 168) || + first >= 224 + ) + } + + if (net.isIPv6(address)) { + const normalized = address.toLowerCase() + return ( + normalized === '::1' || + normalized === '::' || + normalized.startsWith('fc') || + normalized.startsWith('fd') || + normalized.startsWith('fe80:') || + normalized.startsWith('ff') + ) + } + + return true +} + +async function validatePushEndpoint(endpoint: string): Promise { + const parsed = new URL(endpoint) + + if (parsed.protocol !== 'https:') { + throw new z.ZodError([ + { + code: z.ZodIssueCode.custom, + path: ['endpoint'], + message: 'Push endpoint must use HTTPS', + }, + ]) + } + + if (parsed.username || parsed.password) { + throw new z.ZodError([ + { + code: z.ZodIssueCode.custom, + path: ['endpoint'], + message: 'Push endpoint must not include credentials', + }, + ]) + } + + const hostname = parsed.hostname.toLowerCase() + if ( + hostname === 'localhost' || + hostname.endsWith('.localhost') || + hostname.endsWith('.local') + ) { + throw new z.ZodError([ + { + code: z.ZodIssueCode.custom, + path: ['endpoint'], + message: 'Push endpoint host is not allowed', + }, + ]) + } + + if (net.isIP(hostname)) { + if (isBlockedIpAddress(hostname)) { + throw new z.ZodError([ + { + code: z.ZodIssueCode.custom, + path: ['endpoint'], + message: 'Push endpoint IP range is not allowed', + }, + ]) + } + return + } + + const resolved = await dns.lookup(hostname, { all: true, verbatim: false }) + if (resolved.some(entry => isBlockedIpAddress(entry.address))) { + throw new z.ZodError([ + { + code: z.ZodIssueCode.custom, + path: ['endpoint'], + message: 'Push endpoint resolves to a disallowed IP range', + }, + ]) + } +} + +export async function pushRoutes(fastify: FastifyInstance) { + fastify.get('/vapid-public-key', async (_request, reply) => { + if (!config.webPush.enabled || !config.webPush.publicKey) { + return reply.status(501).send({ + success: false, + error: { + code: 'PUSH_NOT_CONFIGURED', + message: 'Push notifications are not configured', + }, + meta: { + timestamp: new Date().toISOString(), + }, + }) + } + + const response: APIResponse<{ publicKey: string }> = { + success: true, + data: { + publicKey: config.webPush.publicKey, + }, + meta: { + timestamp: new Date().toISOString(), + version: '1.0.0', + }, + } + + return reply.send(response) + }) +} + +export async function userPushSubscriptionRoutes(fastify: FastifyInstance) { + const pool = getPostgresPool() + + fastify.get('/', { preHandler: requireAuth }, async (request, reply) => { + if (!request.user) { + return reply.status(401).send({ + success: false, + error: { + code: 'UNAUTHORIZED', + message: 'Not authenticated', + }, + meta: { + timestamp: new Date().toISOString(), + }, + }) + } + + try { + const userId = Number(request.user.sub) + const result = await pool.query( + `SELECT id, endpoint, device_name, enabled, last_seen_at, created_at + FROM push_subscriptions + WHERE user_id = $1 + ORDER BY last_seen_at DESC`, + [userId], + ) + + const response: APIResponse<{ + subscriptions: ReturnType[] + }> = { + success: true, + data: { + subscriptions: result.rows.map(serializeSubscription), + }, + meta: { + timestamp: new Date().toISOString(), + version: '1.0.0', + }, + } + + return reply.send(response) + } catch (error) { + fastify.log.error({ error }, 'Error fetching push subscriptions') + return reply.status(500).send({ + success: false, + error: { + code: 'INTERNAL_ERROR', + message: 'Failed to fetch push subscriptions', + }, + meta: { + timestamp: new Date().toISOString(), + }, + }) + } + }) + + fastify.post('/', { preHandler: requireAuth }, async (request, reply) => { + if (!request.user) { + return reply.status(401).send({ + success: false, + error: { + code: 'UNAUTHORIZED', + message: 'Not authenticated', + }, + meta: { + timestamp: new Date().toISOString(), + }, + }) + } + + try { + const userId = Number(request.user.sub) + const body = PushSubscriptionBodySchema.parse(request.body) + await validatePushEndpoint(body.endpoint) + + const client = await pool.connect() + + try { + await client.query('BEGIN') + await client.query('SELECT pg_advisory_xact_lock($1)', [userId]) + + const existingResult = await client.query<{ id: number }>( + 'SELECT id FROM push_subscriptions WHERE user_id = $1 AND endpoint = $2', + [userId, body.endpoint], + ) + + if (existingResult.rows.length === 0) { + const countResult = await client.query<{ count: string }>( + 'SELECT COUNT(*)::text AS count FROM push_subscriptions WHERE user_id = $1 AND enabled = TRUE', + [userId], + ) + const activeCount = Number( + countResult.rows[0]?.count || '0', + ) + + if (activeCount >= MAX_PUSH_SUBSCRIPTIONS_PER_USER) { + await client.query('ROLLBACK') + return reply.status(409).send({ + success: false, + error: { + code: 'SUBSCRIPTION_LIMIT_REACHED', + message: `You can register up to ${MAX_PUSH_SUBSCRIPTIONS_PER_USER} push subscriptions`, + }, + meta: { + timestamp: new Date().toISOString(), + }, + }) + } + } + + const userAgent = getUserAgent(request.headers['user-agent']) + const result = await client.query( + `INSERT INTO push_subscriptions ( + user_id, endpoint, p256dh, auth, expiration_time, device_name, user_agent, enabled, last_seen_at + ) VALUES ($1, $2, $3, $4, $5, $6, $7, TRUE, NOW()) + ON CONFLICT (endpoint) DO UPDATE SET + user_id = EXCLUDED.user_id, + p256dh = EXCLUDED.p256dh, + auth = EXCLUDED.auth, + expiration_time = EXCLUDED.expiration_time, + device_name = EXCLUDED.device_name, + user_agent = EXCLUDED.user_agent, + enabled = TRUE, + last_seen_at = NOW() + RETURNING id, endpoint, device_name, enabled, last_seen_at, created_at`, + [ + userId, + body.endpoint, + body.keys.p256dh, + body.keys.auth, + parseExpirationTime(body.expirationTime), + body.deviceName || null, + userAgent, + ], + ) + + await client.query('COMMIT') + + const response: APIResponse<{ + subscription: ReturnType + }> = { + success: true, + data: { + subscription: serializeSubscription(result.rows[0]), + }, + meta: { + timestamp: new Date().toISOString(), + version: '1.0.0', + }, + } + + return reply + .status(existingResult.rows.length === 0 ? 201 : 200) + .send(response) + } catch (error) { + await client.query('ROLLBACK') + throw error + } finally { + client.release() + } + } catch (error) { + fastify.log.error({ error }, 'Error upserting push subscription') + + if (error instanceof z.ZodError) { + return reply.status(400).send({ + success: false, + error: { + code: 'VALIDATION_ERROR', + message: 'Invalid push subscription payload', + details: error.errors, + }, + meta: { + timestamp: new Date().toISOString(), + }, + }) + } + + return reply.status(500).send({ + success: false, + error: { + code: 'INTERNAL_ERROR', + message: 'Failed to save push subscription', + }, + meta: { + timestamp: new Date().toISOString(), + }, + }) + } + }) + + fastify.delete( + '/:id', + { preHandler: requireAuth }, + async (request, reply) => { + if (!request.user) { + return reply.status(401).send({ + success: false, + error: { + code: 'UNAUTHORIZED', + message: 'Not authenticated', + }, + meta: { + timestamp: new Date().toISOString(), + }, + }) + } + + try { + const userId = Number(request.user.sub) + const { id } = SubscriptionParamsSchema.parse(request.params) + const result = await pool.query<{ id: number }>( + 'DELETE FROM push_subscriptions WHERE id = $1 AND user_id = $2 RETURNING id', + [id, userId], + ) + + if (result.rows.length === 0) { + return reply.status(404).send({ + success: false, + error: { + code: 'NOT_FOUND', + message: 'Push subscription not found', + }, + meta: { + timestamp: new Date().toISOString(), + }, + }) + } + + const response: APIResponse = { + success: true, + data: null, + meta: { + timestamp: new Date().toISOString(), + version: '1.0.0', + }, + } + + return reply.send(response) + } catch (error) { + fastify.log.error({ error }, 'Error deleting push subscription') + + if (error instanceof z.ZodError) { + return reply.status(400).send({ + success: false, + error: { + code: 'VALIDATION_ERROR', + message: 'Invalid push subscription id', + details: error.errors, + }, + meta: { + timestamp: new Date().toISOString(), + }, + }) + } + + return reply.status(500).send({ + success: false, + error: { + code: 'INTERNAL_ERROR', + message: 'Failed to delete push subscription', + }, + meta: { + timestamp: new Date().toISOString(), + }, + }) + } + }, + ) +} diff --git a/services/shared/src/config/index.ts b/services/shared/src/config/index.ts index 0123fc8..ceb0c72 100644 --- a/services/shared/src/config/index.ts +++ b/services/shared/src/config/index.ts @@ -68,6 +68,13 @@ const ConfigSchema = z.object({ fromEmail: z.string().default('noreply@grails.market'), enabled: z.boolean().default(true), }), + webPush: z.object({ + publicKey: z.string().optional(), + privateKey: z.string().optional(), + subject: z.string().default('mailto:noreply@grails.market'), + ttlSeconds: z.number().int().positive().default(86400), + enabled: z.boolean(), + }), frontend: z.object({ url: z.string().default('http://localhost:3001'), }), @@ -170,6 +177,13 @@ const rawConfig = { fromEmail: process.env.FROM_EMAIL, enabled: process.env.ENABLE_EMAIL !== 'false', }, + webPush: { + publicKey: process.env.WEB_PUSH_VAPID_PUBLIC_KEY, + privateKey: process.env.WEB_PUSH_VAPID_PRIVATE_KEY, + subject: process.env.WEB_PUSH_SUBJECT || `mailto:${process.env.FROM_EMAIL || 'noreply@grails.market'}`, + ttlSeconds: process.env.WEB_PUSH_TTL_SECONDS ? parseInt(process.env.WEB_PUSH_TTL_SECONDS) : 86400, + enabled: !!(process.env.WEB_PUSH_VAPID_PUBLIC_KEY && process.env.WEB_PUSH_VAPID_PRIVATE_KEY), + }, frontend: { url: process.env.FRONTEND_URL, }, @@ -265,4 +279,4 @@ export function getRegistrationSource(referrer: string): string | null { return ENS_REFERRER_CODES[referrer.toLowerCase()] ?? null; } -export default config; \ No newline at end of file +export default config; diff --git a/services/shared/src/db/schema.sql b/services/shared/src/db/schema.sql index f032970..9630ccb 100644 --- a/services/shared/src/db/schema.sql +++ b/services/shared/src/db/schema.sql @@ -88,6 +88,43 @@ CREATE INDEX IF NOT EXISTS idx_ens_names_owner ON ens_names(owner_address); CREATE INDEX IF NOT EXISTS idx_ens_names_expiry ON ens_names(expiry_date); CREATE INDEX IF NOT EXISTS idx_ens_names_name_lower ON ens_names(LOWER(name)); +-- Browser Web Push subscriptions. This mirrors API migration +-- 0894_create_push_subscriptions.sql for schema-based bootstrap flows. The +-- conditional block keeps this legacy schema file usable in environments where +-- the users table is supplied by the API migration sequence instead. +DO $$ +BEGIN + IF EXISTS ( + SELECT 1 FROM information_schema.tables + WHERE table_schema = 'public' AND table_name = 'users' + ) THEN + CREATE TABLE IF NOT EXISTS push_subscriptions ( + id SERIAL PRIMARY KEY, + user_id INTEGER NOT NULL REFERENCES users(id) ON DELETE CASCADE, + endpoint TEXT NOT NULL UNIQUE, + p256dh TEXT NOT NULL, + auth TEXT NOT NULL, + expiration_time TIMESTAMP, + device_name VARCHAR(120), + user_agent TEXT, + enabled BOOLEAN NOT NULL DEFAULT TRUE, + last_seen_at TIMESTAMP NOT NULL DEFAULT NOW(), + created_at TIMESTAMP NOT NULL DEFAULT NOW(), + updated_at TIMESTAMP NOT NULL DEFAULT NOW() + ); + + CREATE INDEX IF NOT EXISTS idx_push_subscriptions_user + ON push_subscriptions(user_id); + + CREATE INDEX IF NOT EXISTS idx_push_subscriptions_enabled_user + ON push_subscriptions(user_id) + WHERE enabled = TRUE; + + CREATE INDEX IF NOT EXISTS idx_push_subscriptions_endpoint + ON push_subscriptions(endpoint); + END IF; +END $$; + CREATE INDEX IF NOT EXISTS idx_listings_status ON listings(status); CREATE INDEX IF NOT EXISTS idx_listings_seller ON listings(seller_address); CREATE INDEX IF NOT EXISTS idx_listings_created ON listings(created_at DESC); @@ -128,6 +165,18 @@ DROP TRIGGER IF EXISTS update_indexer_state_updated_at ON indexer_state; CREATE TRIGGER update_indexer_state_updated_at BEFORE UPDATE ON indexer_state FOR EACH ROW EXECUTE FUNCTION update_updated_at_column(); +DO $$ +BEGIN + IF EXISTS ( + SELECT 1 FROM information_schema.tables + WHERE table_schema = 'public' AND table_name = 'push_subscriptions' + ) THEN + DROP TRIGGER IF EXISTS update_push_subscriptions_updated_at ON push_subscriptions; + CREATE TRIGGER update_push_subscriptions_updated_at BEFORE UPDATE ON push_subscriptions + FOR EACH ROW EXECUTE FUNCTION update_updated_at_column(); + END IF; +END $$; + -- Enable logical replication for WAL listener -- Note: These commands need to be run as superuser -- ALTER SYSTEM SET wal_level = logical; @@ -529,4 +578,4 @@ CREATE TABLE IF NOT EXISTS valuation_settings ( updated_at TIMESTAMP NOT NULL DEFAULT NOW() ); INSERT INTO valuation_settings (id, quota_avatar) VALUES (1, 60) - ON CONFLICT (id) DO NOTHING; \ No newline at end of file + ON CONFLICT (id) DO NOTHING; diff --git a/services/workers/package.json b/services/workers/package.json index 7dff17b..3b75dfb 100644 --- a/services/workers/package.json +++ b/services/workers/package.json @@ -17,11 +17,13 @@ "nodemailer": "^7.0.9", "pg-boss": "^9.0.3", "pino": "^8.17.2", - "pino-pretty": "^10.3.1" + "pino-pretty": "^10.3.1", + "web-push": "^3.6.7" }, "devDependencies": { "@types/node": "^20.10.0", "@types/nodemailer": "^7.0.2", + "@types/web-push": "^3.6.4", "tsx": "^4.7.0", "typescript": "^5.3.0" } diff --git a/services/workers/src/queue.ts b/services/workers/src/queue.ts index 0c8fb69..9553915 100644 --- a/services/workers/src/queue.ts +++ b/services/workers/src/queue.ts @@ -89,7 +89,8 @@ export interface SendNotificationJob { | 'listing-cancelled-ownership-change' | 'offer-received' | 'listing-sold' - | 'comment-received'; + | 'comment-received' + | 'listing-cancelled'; userId?: number; email?: string; recipientAddress?: string; diff --git a/services/workers/src/utils/push-notification.ts b/services/workers/src/utils/push-notification.ts new file mode 100644 index 0000000..d8d48ee --- /dev/null +++ b/services/workers/src/utils/push-notification.ts @@ -0,0 +1,151 @@ +import { getPostgresPool, config } from '../../../shared/src'; +import { logger } from '../utils/logger'; +import type { SendNotificationJob } from '../queue'; +import webPush from 'web-push'; + +const FRONTEND_URL = config.frontend.url; + +interface PushSubscriptionRow { + id: number; + endpoint: string; + p256dh: string; + auth: string; +} + +function configureWebPush(): boolean { + if (!config.webPush.enabled || !config.webPush.publicKey || !config.webPush.privateKey) { + return false; + } + + webPush.setVapidDetails(config.webPush.subject, config.webPush.publicKey, config.webPush.privateKey); + + return true; +} + +function buildPushBody( + type: SendNotificationJob['type'], + ensName: string, + metadata: Record | undefined, +): string { + switch (type) { + case 'new-listing': + return `${ensName} was listed${typeof metadata?.priceWei === 'string' ? ' on Grails' : ''}`; + case 'price-change': + return `${ensName} has a new listing price`; + case 'sale': + return `${ensName} sold`; + case 'new-offer': + case 'offer-received': + return `${ensName} received a new offer`; + case 'listing-sold': + return `Your listing for ${ensName} sold`; + case 'listing-cancelled-ownership-change': + case 'listing-cancelled': + return `The listing for ${ensName} was cancelled`; + case 'comment-received': + return `${ensName} received a new comment`; + } +} + +function buildPushTitle(type: SendNotificationJob['type']): string { + switch (type) { + case 'new-listing': + return 'New listing'; + case 'price-change': + return 'Price changed'; + case 'sale': + return 'Name sold'; + case 'new-offer': + case 'offer-received': + return 'New offer'; + case 'listing-sold': + return 'Listing sold'; + case 'listing-cancelled-ownership-change': + case 'listing-cancelled': + return 'Listing cancelled'; + case 'comment-received': + return 'New comment'; + } +} + +export async function sendPushNotifications(params: { + userId: number; + type: SendNotificationJob['type']; + ensName: string; + notificationId: number; + metadata?: Record; +}): Promise { + if (!configureWebPush()) { + logger.debug({ userId: params.userId }, 'Web Push not configured, skipping push delivery'); + return; + } + + const pool = getPostgresPool(); + const subscriptionsResult = await pool.query( + `SELECT id, endpoint, p256dh, auth + FROM push_subscriptions + WHERE user_id = $1 + AND enabled = TRUE + AND (expiration_time IS NULL OR expiration_time > NOW())`, + [params.userId], + ); + + if (subscriptionsResult.rows.length === 0) { + logger.debug({ userId: params.userId }, 'No active push subscriptions for user'); + return; + } + + const payload = JSON.stringify({ + title: buildPushTitle(params.type), + body: buildPushBody(params.type, params.ensName, params.metadata), + url: `${FRONTEND_URL}/${params.ensName}`, + notificationId: params.notificationId, + type: params.type, + metadata: params.metadata || {}, + }); + + const results = await Promise.allSettled( + subscriptionsResult.rows.map(async (subscriptionRow: PushSubscriptionRow) => { + const subscription: webPush.PushSubscription = { + endpoint: subscriptionRow.endpoint, + keys: { + p256dh: subscriptionRow.p256dh, + auth: subscriptionRow.auth, + }, + }; + + try { + await webPush.sendNotification(subscription, payload, { + TTL: config.webPush.ttlSeconds, + }); + } catch (error) { + if (error instanceof webPush.WebPushError && (error.statusCode === 404 || error.statusCode === 410)) { + await pool.query('DELETE FROM push_subscriptions WHERE id = $1', [subscriptionRow.id]); + logger.info( + { + userId: params.userId, + subscriptionId: subscriptionRow.id, + statusCode: error.statusCode, + }, + 'Removed stale push subscription', + ); + return; + } + + throw error; + } + }), + ); + + const failedCount = results.filter((result: PromiseSettledResult) => result.status === 'rejected').length; + if (failedCount > 0) { + logger.warn( + { + userId: params.userId, + failedCount, + total: subscriptionsResult.rows.length, + }, + 'Some push deliveries failed', + ); + } +} diff --git a/services/workers/src/workers/notifications.ts b/services/workers/src/workers/notifications.ts index 0871e56..74d26fc 100644 --- a/services/workers/src/workers/notifications.ts +++ b/services/workers/src/workers/notifications.ts @@ -1,7 +1,9 @@ import PgBoss from 'pg-boss'; -import { getPostgresPool, config } from '../../../shared/src'; +import { ethers } from 'ethers'; import { logger } from '../utils/logger'; +import { getPostgresPool, config } from '../../../shared/src'; import { QUEUE_NAMES, type SendNotificationJob } from '../queue'; +import { sendPushNotifications } from '../utils/push-notification'; import { sendEmail, buildNewListingEmail, @@ -12,11 +14,42 @@ import { buildOfferReceivedEmail, buildListingSoldEmail, buildCommentReceivedEmail, + type EmailTemplate, } from '../services/email'; -import { ethers } from 'ethers'; const FRONTEND_URL = config.frontend.url; +interface EnsNameRow { + name: string; +} + +interface UserNotificationRow { + email: string | null; + email_verified: boolean; +} + +interface ExistingNotificationRow { + id: number; + metadata: Record | null; +} + +interface InsertedNotificationRow { + id: number; +} + +function metadataString(metadata: Record | undefined, key: string): string { + const value = metadata?.[key]; + if (typeof value === 'string') { + return value; + } + + if (typeof value === 'number' || typeof value === 'bigint') { + return value.toString(); + } + + return '0'; +} + /** * Notification Worker * @@ -31,7 +64,7 @@ export async function registerNotificationWorker(boss: PgBoss): Promise { teamConcurrency: 2, }, async (job) => { - const { type, userId, email, recipientAddress, ensNameId, metadata, transactionHash } = job.data; + const { type, userId, email, recipientAddress, ensNameId, metadata } = job.data; logger.info({ type, userId, ensNameId }, 'Processing notification'); @@ -39,10 +72,7 @@ export async function registerNotificationWorker(boss: PgBoss): Promise { try { // Get ENS name details - const ensResult = await pool.query( - 'SELECT name FROM ens_names WHERE id = $1', - [ensNameId] - ); + const ensResult = await pool.query('SELECT name FROM ens_names WHERE id = $1', [ensNameId]); if (ensResult.rows.length === 0) { logger.warn({ ensNameId }, 'ENS name not found for notification'); @@ -52,11 +82,11 @@ export async function registerNotificationWorker(boss: PgBoss): Promise { const ensName = ensResult.rows[0].name; // Get recipient email if not provided - let recipientEmail = email; + let recipientEmail: string | null | undefined = email; if (!recipientEmail && userId) { - const userResult = await pool.query( + const userResult = await pool.query( 'SELECT email, email_verified FROM users WHERE id = $1', - [userId] + [userId], ); if (userResult.rows.length === 0) { @@ -66,7 +96,6 @@ export async function registerNotificationWorker(boss: PgBoss): Promise { const user = userResult.rows[0]; - // Check if email is verified if (!user.email_verified) { logger.info({ userId }, 'User email not verified, skipping notification'); return; @@ -83,8 +112,7 @@ export async function registerNotificationWorker(boss: PgBoss): Promise { } if (!recipientEmail) { - logger.warn({ type, userId, ensNameId }, 'No email found for notification recipient'); - return; + logger.info({ type, userId, ensNameId }, 'No verified email found for notification recipient'); } // Check if we already sent this notification (deduplication) @@ -96,13 +124,13 @@ export async function registerNotificationWorker(boss: PgBoss): Promise { const allowDuplicates = ['listing-sold', 'comment-received']; if (!allowDuplicates.includes(type)) { - const existingNotification = await pool.query( + const existingNotification = await pool.query( `SELECT id, metadata FROM notifications WHERE user_id = $1 AND type = $2 AND ens_name_id = $3 AND sent_at > NOW() - INTERVAL '12 hours'`, - [userId, type, ensNameId] + [userId, type, ensNameId], ); if (existingNotification.rows.length > 0) { @@ -111,20 +139,26 @@ export async function registerNotificationWorker(boss: PgBoss): Promise { const lastNotificationPrice = existingNotification.rows[0].metadata?.priceWei; if (lastNotificationPrice && lastNotificationPrice !== metadata.priceWei) { logger.info( - { userId, type, ensNameId, oldPrice: lastNotificationPrice, newPrice: metadata.priceWei }, - 'Price changed since last notification, allowing duplicate' + { + userId, + type, + ensNameId, + oldPrice: lastNotificationPrice, + newPrice: metadata.priceWei, + }, + 'Price changed since last notification, allowing duplicate', ); } else { logger.info( { userId, type, ensNameId }, - 'Duplicate notification detected (sent within last 12 hours), skipping' + 'Duplicate notification detected (sent within last 12 hours), skipping', ); return; } } else { logger.info( { userId, type, ensNameId }, - 'Duplicate notification detected (sent within last 12 hours), skipping' + 'Duplicate notification detected (sent within last 12 hours), skipping', ); return; } @@ -133,12 +167,12 @@ export async function registerNotificationWorker(boss: PgBoss): Promise { } // Build email based on notification type - let emailTemplate; + let emailTemplate: EmailTemplate | undefined; const unsubscribeUrl = `${FRONTEND_URL}/settings/notifications`; switch (type) { case 'new-listing': { - const priceWei = metadata?.priceWei || '0'; + const priceWei = metadataString(metadata, 'priceWei'); const priceEth = ethers.formatEther(priceWei); emailTemplate = buildNewListingEmail({ @@ -151,8 +185,8 @@ export async function registerNotificationWorker(boss: PgBoss): Promise { } case 'price-change': { - const oldPriceWei = metadata?.oldPriceWei || '0'; - const newPriceWei = metadata?.newPriceWei || '0'; + const oldPriceWei = metadataString(metadata, 'oldPriceWei'); + const newPriceWei = metadataString(metadata, 'newPriceWei'); const oldPriceEth = ethers.formatEther(oldPriceWei); const newPriceEth = ethers.formatEther(newPriceWei); @@ -167,7 +201,7 @@ export async function registerNotificationWorker(boss: PgBoss): Promise { } case 'sale': { - const priceWei = metadata?.priceWei || '0'; + const priceWei = metadataString(metadata, 'priceWei'); const priceEth = ethers.formatEther(priceWei); emailTemplate = buildSaleEmail({ @@ -180,7 +214,7 @@ export async function registerNotificationWorker(boss: PgBoss): Promise { } case 'new-offer': { - const offerAmountWei = metadata?.offerAmountWei || '0'; + const offerAmountWei = metadataString(metadata, 'offerAmountWei'); const priceEth = ethers.formatEther(offerAmountWei); emailTemplate = buildNewOfferEmail({ @@ -201,8 +235,17 @@ export async function registerNotificationWorker(boss: PgBoss): Promise { break; } + case 'listing-cancelled': { + emailTemplate = buildListingCancelledEmail({ + ensName, + listingUrl: `${FRONTEND_URL}/${ensName}`, + unsubscribeUrl, + }); + break; + } + case 'offer-received': { - const offerAmountWei = metadata?.offerAmountWei || '0'; + const offerAmountWei = metadataString(metadata, 'offerAmountWei'); const priceEth = ethers.formatEther(offerAmountWei); emailTemplate = buildOfferReceivedEmail({ @@ -215,7 +258,7 @@ export async function registerNotificationWorker(boss: PgBoss): Promise { } case 'listing-sold': { - const priceWei = metadata?.priceWei || '0'; + const priceWei = metadataString(metadata, 'priceWei'); const priceEth = ethers.formatEther(priceWei); emailTemplate = buildListingSoldEmail({ @@ -241,16 +284,50 @@ export async function registerNotificationWorker(boss: PgBoss): Promise { return; } - // Send email - await sendEmail(recipientEmail, emailTemplate); - - // Log notification in database + // Log notification in database as the canonical in-app notification + let notificationId: number | undefined; if (userId) { - await pool.query( + const insertedNotification = await pool.query( `INSERT INTO notifications (user_id, type, ens_name_id, metadata, sent_at) - VALUES ($1, $2, $3, $4, NOW())`, - [userId, type, ensNameId, JSON.stringify(metadata || {})] + VALUES ($1, $2, $3, $4, NOW()) + RETURNING id`, + [userId, type, ensNameId, JSON.stringify(metadata || {})], ); + notificationId = insertedNotification.rows[0].id; + } + + if (recipientEmail && emailTemplate) { + try { + await sendEmail(recipientEmail, emailTemplate); + } catch (emailError) { + logger.warn( + { + error: emailError, + userId, + type, + ensNameId, + email: recipientEmail, + }, + 'Email notification delivery failed after canonical notification was logged', + ); + } + } + + if (userId && notificationId !== undefined) { + try { + await sendPushNotifications({ + userId, + type, + ensName, + notificationId, + metadata, + }); + } catch (pushError) { + logger.warn( + { error: pushError, userId, type, ensNameId }, + 'Push notification delivery failed after canonical notification was logged', + ); + } logger.info({ userId, type, ensNameId, email: recipientEmail }, 'Notification sent and logged'); } else { @@ -260,7 +337,7 @@ export async function registerNotificationWorker(boss: PgBoss): Promise { logger.error({ error, type, userId, ensNameId }, 'Error sending notification'); throw error; // Will trigger pg-boss retry } - } + }, ); logger.info('Notification worker registered'); diff --git a/services/workers/src/workers/ownership.ts b/services/workers/src/workers/ownership.ts index 7825f40..edba986 100644 --- a/services/workers/src/workers/ownership.ts +++ b/services/workers/src/workers/ownership.ts @@ -1,7 +1,18 @@ import PgBoss from 'pg-boss'; import { getPostgresPool } from '../../../shared/src'; import { logger } from '../utils/logger'; -import { QUEUE_NAMES, type UpdateOwnershipJob } from '../queue'; +import { QUEUE_NAMES, type SendNotificationJob, type UpdateOwnershipJob } from '../queue'; + +interface UnfundedListingRow { + id: number; + seller_address: string; + price_wei: string; +} + +interface UserRow { + id: number; + address: string; +} /** * Ownership Update Worker @@ -39,6 +50,7 @@ export async function registerOwnershipWorker(boss: PgBoss): Promise { const pool = getPostgresPool(); const client = await pool.connect(); + const notificationJobs: SendNotificationJob[] = []; try { await client.query('BEGIN'); @@ -105,7 +117,7 @@ export async function registerOwnershipWorker(boss: PgBoss): Promise { ); // Trigger immediate validation for these listings with singletonKey to prevent duplicates - const validationJobs = unfundedListings.rows.map((listing) => ({ + const validationJobs = unfundedListings.rows.map((listing: UnfundedListingRow) => ({ name: 'validate-listing-ownership', data: { listingId: listing.id }, singletonKey: `listing-${listing.id}` @@ -116,9 +128,48 @@ export async function registerOwnershipWorker(boss: PgBoss): Promise { { count: validationJobs.length }, 'Validation jobs queued for unfunded listings' ); + + const sellerAddresses = Array.from( + new Set(unfundedListings.rows.map((listing: UnfundedListingRow) => listing.seller_address.toLowerCase())) + ); + const usersResult = await client.query( + 'SELECT id, address FROM users WHERE LOWER(address) = ANY($1::text[])', + [sellerAddresses] + ); + const usersByAddress = new Map( + usersResult.rows.map((user: UserRow) => [user.address.toLowerCase(), user.id]) + ); + + for (const listing of unfundedListings.rows as UnfundedListingRow[]) { + const userId = usersByAddress.get(listing.seller_address.toLowerCase()); + if (userId !== undefined) { + notificationJobs.push({ + type: 'listing-cancelled', + userId, + ensNameId, + metadata: { + listingId: listing.id, + priceWei: listing.price_wei, + reason: 'ownership_lost', + transactionHash, + }, + }); + } + } } await client.query('COMMIT'); + + if (notificationJobs.length > 0) { + await Promise.all( + notificationJobs.map((notificationJob) => boss.send(QUEUE_NAMES.SEND_NOTIFICATION, notificationJob)) + ); + logger.info( + { count: notificationJobs.length, ensNameId, ensName }, + 'Notification jobs queued for unfunded listings' + ); + } + logger.info({ ensNameId, ensName }, 'Ownership update transaction completed'); } catch (error) { await client.query('ROLLBACK');