push notifications - #230
Conversation
global chat
valuations
valuations fixes
chat features
valuation stream refactor
chat images
val props and cache
chat multi images
Ultraworked with [Sisyphus](https://github.com/code-yeongyu/oh-my-openagent)
Ultraworked with [Sisyphus](https://github.com/code-yeongyu/oh-my-openagent)
Ultraworked with [Sisyphus](https://github.com/code-yeongyu/oh-my-openagent)
Ultraworked with [Sisyphus](https://github.com/code-yeongyu/oh-my-openagent)
Greptile SummaryThis PR introduces browser Web Push notification delivery: a new
Confidence Score: 3/5The push subscription upsert allows any authenticated user to claim another user's endpoint, redirecting that user's future push notifications. This needs to be resolved before the feature goes live. The subscription upsert unconditionally overwrites the owner of any endpoint on conflict without checking whether the endpoint already belongs to a different user. A malicious or mistaken client could silently redirect another user's push notifications to their own account. The DNS validation gap is a secondary concern. The rest of the changes — migration, config, ownership worker extension — are straightforward and low-risk. services/api/src/routes/push.ts deserves the most scrutiny; services/workers/src/workers/notifications.ts should have its VAPID initialization path reviewed.
|
| Filename | Overview |
|---|---|
| services/api/src/routes/push.ts | New 431-line route file adding VAPID key endpoint + 3 authenticated subscription CRUD handlers; contains an upsert that silently transfers push subscription ownership across users and a DNS validation gap before storing the endpoint. |
| services/workers/src/workers/notifications.ts | Adds ~170 lines of push-notification fan-out after canonical notification insertion; VAPID global state is re-initialized on every delivery rather than once at startup, and the new-listing push body condition is always true. |
| services/workers/src/workers/ownership.ts | Adds listing-cancelled notification jobs queued after the ownership-update transaction commits; correct transaction boundary and ensName is properly in scope. |
| services/api/migrations/seq/0894_create_push_subscriptions.sql | Clean migration creating push_subscriptions table with correct FK cascade, unique endpoint constraint, partial index on enabled=TRUE, and an updated_at trigger. |
| services/shared/src/config/index.ts | Adds webPush config block; ttlSeconds default is duplicated between rawConfig and the Zod schema, harmless. |
| services/workers/src/queue.ts | Adds listing-cancelled notification type and tightens metadata type from Record<string,any> to Record<string,unknown>; clean change. |
Sequence Diagram
%%{init: {'theme': 'neutral'}}%%
sequenceDiagram
participant Browser
participant API as API (push.ts)
participant DB as PostgreSQL
participant Worker as Notification Worker
participant PushSvc as Push Service (FCM/Mozilla)
Browser->>API: GET /api/v1/push/vapid-public-key
API-->>Browser: publicKey (or 501 if unconfigured)
Browser->>API: POST /api/v1/users/me/push-subscriptions
Note over API: validatePushEndpoint (DNS check)
API->>DB: pg_advisory_xact_lock(userId)
API->>DB: SELECT existing by user_id + endpoint
API->>DB: SELECT count for user
API->>DB: "INSERT ... ON CONFLICT (endpoint) DO UPDATE SET user_id=EXCLUDED.user_id"
DB-->>API: subscription row
API-->>Browser: 201/200 subscription
Note over Worker: On ownership change or listing event
Worker->>DB: INSERT notification RETURNING id
Worker->>DB: SELECT push_subscriptions WHERE user_id AND enabled
loop Each subscription
Worker->>PushSvc: sendNotification (VAPID-signed)
alt 404/410
Worker->>DB: DELETE stale subscription
end
end
Browser->>API: DELETE /api/v1/users/me/push-subscriptions/:id
API->>DB: DELETE WHERE id AND user_id
API-->>Browser: 200 / 404
%%{init: {'theme': 'base', 'themeVariables': {"darkMode": true, "background": "#0d1117", "primaryColor": "#21262d", "primaryTextColor": "#e6edf3", "primaryBorderColor": "#8b949e", "lineColor": "#8b949e", "textColor": "#e6edf3", "edgeLabelBackground": "#161b22", "actorBkg": "#21262d", "actorBorder": "#8b949e", "actorTextColor": "#e6edf3", "actorLineColor": "#8b949e", "signalColor": "#8b949e", "signalTextColor": "#e6edf3", "noteBkgColor": "#373320", "noteBorderColor": "#d4a72c", "noteTextColor": "#f0e6c0", "labelBoxBkgColor": "#21262d", "labelBoxBorderColor": "#8b949e", "labelTextColor": "#e6edf3", "loopTextColor": "#e6edf3", "activationBkgColor": "#30363d", "activationBorderColor": "#8b949e"}}}%%
sequenceDiagram
participant Browser
participant API as API (push.ts)
participant DB as PostgreSQL
participant Worker as Notification Worker
participant PushSvc as Push Service (FCM/Mozilla)
Browser->>API: GET /api/v1/push/vapid-public-key
API-->>Browser: publicKey (or 501 if unconfigured)
Browser->>API: POST /api/v1/users/me/push-subscriptions
Note over API: validatePushEndpoint (DNS check)
API->>DB: pg_advisory_xact_lock(userId)
API->>DB: SELECT existing by user_id + endpoint
API->>DB: SELECT count for user
API->>DB: "INSERT ... ON CONFLICT (endpoint) DO UPDATE SET user_id=EXCLUDED.user_id"
DB-->>API: subscription row
API-->>Browser: 201/200 subscription
Note over Worker: On ownership change or listing event
Worker->>DB: INSERT notification RETURNING id
Worker->>DB: SELECT push_subscriptions WHERE user_id AND enabled
loop Each subscription
Worker->>PushSvc: sendNotification (VAPID-signed)
alt 404/410
Worker->>DB: DELETE stale subscription
end
end
Browser->>API: DELETE /api/v1/users/me/push-subscriptions/:id
API->>DB: DELETE WHERE id AND user_id
API-->>Browser: 200 / 404
Reviews (1): Last reviewed commit: "ownership notification jobs" | Re-trigger Greptile
![greptile-apps[bot]](https://avatars.githubusercontent.com/in/867647?s=60&v=4){kind=small}
| `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() |
There was a problem hiding this comment.
user_id = EXCLUDED.user_id in the conflict clause overwrites the user_id of any existing subscription, even one that belongs to a different user. The pre-check at L254-257 (WHERE user_id = $1 AND endpoint = $2) only checks whether the current user already owns the endpoint — it returns 0 rows when the endpoint belongs to user A, so the count guard passes and the upsert fires, moving the endpoint from user A to user B. After that, push notifications intended for user A are silently delivered to user B's browser. The intentional device-reuse path (same device, different login) would work correctly if you simply refused to update user_id on conflict: the previous owner's subscription stays in place, and the new user gets a fresh record (or you surface the conflict as a clear signal to revoke the old one first).
| 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; | ||
| } |
There was a problem hiding this comment.
configureWebPush() calls webPush.setVapidDetails() inside sendPushNotifications(), which is invoked once per notification job. setVapidDetails mutates module-level state in the web-push library. With teamConcurrency: 2 two worker goroutines can interleave these calls, and while Node's event loop keeps each JS call atomic, the function has no idempotency guard and performs unnecessary I/O-adjacent work on the hot path. The VAPID keys are stable config values — call webPush.setVapidDetails once during registerNotificationWorker startup (or even at module load) and remove configureWebPush from the per-notification path.
| 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<PushSubscriptionRow>( | ||
| `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<typeof serializeSubscription>[] }> = { | ||
| 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<PushSubscriptionRow>( | ||
| `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<typeof serializeSubscription> }> = { | ||
| 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<null> = { | ||
| 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(), | ||
| }, | ||
| }); | ||
| } | ||
| }); | ||
| } |
There was a problem hiding this comment.
userPushSubscriptionRoutes is 256 lines; the file is 431 lines
The file contains a single large exported function with three inline route handlers, plus DNS/IP helpers and a schema — no logical boundary separates them. The three handlers (GET /, POST /, DELETE /:id) each repeat the same auth-guard pattern and are long enough to be their own files. Consider splitting into push-validation.ts (DNS/IP helpers, schema), push-subscriptions.ts (route handlers), and the existing push.ts for the unauthenticated VAPID key endpoint.
Context Used: Treat any function over 200 lines as a cry for hel... (source)
Note: If this suggestion doesn't match your team's coding style, reply to this and let me know. I'll remember it for next time!
| 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', | ||
| }]); | ||
| } |
There was a problem hiding this comment.
validatePushEndpoint resolves the hostname once at registration time. The resolved IP is not stored, so when web-push later makes the outbound HTTPS request, a DNS rebinding attack can swap the IP to an internal address (e.g. 169.254.169.254) after the validation passes. For most real-world push services (FCM, Mozilla) this is low-risk since their domains are stable, but a malicious or compromised endpoint can exploit the window. The closest reliable mitigation is to store the validated endpoint only after sending a test push and receiving a 201 from the push service, confirming the URL is a real push provider.
| case 'new-listing': | ||
| return `${ensName} was listed${typeof metadata?.priceWei === 'string' ? ' on Grails' : ''}`; |
There was a problem hiding this comment.
new-listing push body condition is always true for real listing events
typeof metadata?.priceWei === 'string' is always true for new-listing jobs because priceWei is always passed as a string in the job metadata. The ternary never takes the '' branch in practice, making the body always read "was listed on Grails" regardless. Either make it unconditional or check for the presence of a meaningful price threshold.
| case 'new-listing': | |
| return `${ensName} was listed${typeof metadata?.priceWei === 'string' ? ' on Grails' : ''}`; | |
| case 'new-listing': | |
| return `${ensName} was listed on Grails`; |
Note: If this suggestion doesn't match your team's coding style, reply to this and let me know. I'll remember it for next time!
|
Opened another PR from a dev branch #231 |
2 participants
Summary