Merge remote-tracking branch 'origin/main' into cron-minutely

This commit is contained in:
mplorentz 2026-09-18 13:11:51 -04:00
commit 8ebee878cb
20 changed files with 494 additions and 30 deletions

4
.beads/.gitignore vendored
View file

@ -71,6 +71,10 @@ backup/
*.db-shm
db.sqlite
bd.db
# Interactions log (runtime, not versioned)
interactions.jsonl
# NOTE: Do NOT add negation patterns here.
# They would override fork protection in .git/info/exclude.
# Config files (metadata.json, config.yaml) are tracked by git by default

View file

@ -52,6 +52,12 @@ COPY --from=build /app/src/emails/ ./dist/emails/
# Create data directory for SQLite
RUN mkdir -p /data
# Create non-root user for security hardening
RUN addgroup -S app && adduser -S -G app app
RUN chown -R app:app /data
USER app
EXPOSE 4738
ENV NODE_ENV=production

View file

@ -1,6 +1,6 @@
# Mailship
A Nostr email notification server. Receives events pushed from relays via [NIP-9a](), stores them, and sends daily or weekly digest emails.
A Nostr email notification server. Receives events pushed from relays via [NIP-9a](https://github.com/nostr-protocol/nips/pull/2194), stores them, and sends daily or weekly digest emails.
This repo is a fork of [anchor](https://github.com/coracle-social/anchor). It has been built primarily to support email notifications in [Flotilla](https://flotilla.social), but supports alternate branding and could be set up to work with any Nostr relay supporting NIP-9a.
@ -79,7 +79,7 @@ Response: { ok: true }
```
### POST /notify/:id
NIP-9a relay push callback. Called by relays or NPB when matching events are found.
[NIP-9a](https://github.com/nostr-protocol/nips/pull/2194) relay push callback. Called by relays or NPB when matching events are found.
```
Body: { id, relay, event? }
@ -112,6 +112,11 @@ Unsubscribe via link from digest email.
## Development
**Requirements:** Node >= 22
> **Single instance:** Mailship uses in-process cron jobs for digest scheduling.
> Running more than one instance concurrently may cause duplicate or missed emails.
```sh
pnpm install
pnpm run build

View file

@ -1,10 +0,0 @@
#!/usr/bin/env bash
# Remove link overrides
node remove-pnpm-overrides.js package.json
# When CI=true as it is on render.com, removing link overrides breaks the lockfile
pnpm i --no-frozen-lockfile
# Build everything
pnpm run build

View file

@ -18,9 +18,9 @@
"@eslint/js": "^9.35.0",
"@types/better-sqlite3": "^7.6.13",
"@types/express": "^5.0.3",
"@types/express-ws": "^3.0.5",
"@types/mjml": "^4.7.4",
"@types/mustache": "^4.2.6",
"@types/node": "^22.18.1",
"@types/nodemailer": "^8.0.1",
"@types/sanitize-html": "^2.16.0",
"@types/ws": "^8.18.1",
@ -34,7 +34,6 @@
"vitest": "^5.0.0"
},
"dependencies": {
"@types/node": "^22.18.1",
"@welshman/content": "^0.6.3",
"@welshman/feeds": "^0.6.3",
"@welshman/lib": "^0.6.3",
@ -43,13 +42,11 @@
"@welshman/signer": "^0.6.3",
"@welshman/store": "^0.6.3",
"@welshman/util": "^0.6.3",
"bcrypt": "^5.1.1",
"cron": "^4.3.3",
"cron-parser": "^5.3.1",
"dotenv": "^16.6.1",
"express": "^4.21.2",
"express-rate-limit": "^7.5.1",
"express-ws": "^5.0.2",
"localstorage-polyfill": "^1.0.1",
"mjml": "^4.15.3",
"mustache": "^4.2.0",
@ -58,12 +55,10 @@
"sanitize-html": "^2.17.0",
"sqlite3": "^5.1.7",
"succinct-async": "^1.0.4",
"ts-node-dev": "^2.0.0",
"ws": "^8.18.3"
},
"pnpm": {
"onlyBuiltDependencies": [
"bcrypt",
"sqlite3"
]
}

Binary file not shown.

View file

@ -41,13 +41,19 @@ export type ConfirmSubscriptionParams = {
export const confirmSubscriptionAction = instrument(
'actions.confirmSubscription',
async ({ token }: ConfirmSubscriptionParams) => {
const sub = await db.confirmSubscription(token)
const result = await db.confirmSubscription(token)
if (!sub) {
if (!result) {
throw new ActionError('That confirmation code is invalid or has expired.')
}
worker.registerSubscription(sub)
// Only register the cron job when this is a fresh confirmation.
// If already confirmed, the job is already running.
if (!result.alreadyConfirmed) {
worker.registerSubscription(result.sub)
}
return result
},
)

View file

@ -195,16 +195,37 @@ export const insertSubscription = instrument(
}
)
export type ConfirmResult = {
sub: Subscription
alreadyConfirmed: boolean
}
export const confirmSubscription = instrument(
'database.confirmSubscription',
async (key: string) => {
return parseSubscription(
async (key: string): Promise<ConfirmResult | undefined> => {
// Try to update an unconfirmed, active row
const updated = parseSubscription(
await get(
`UPDATE subscriptions SET confirmed_at = unixepoch()
WHERE key = ? AND confirmed_at IS NULL AND unsubscribed_at IS NULL RETURNING *`,
[key]
)
)
if (updated) {
return { sub: updated, alreadyConfirmed: false }
}
// No unconfirmed row was updated. Check if the key exists and is
// already confirmed — the user is re-clicking a used link. If the row
// is unsubscribed, treat it as invalid (expired).
const existing = await getSubscriptionByKey(key)
if (!existing || existing.unsubscribed_at) {
return undefined
}
return { sub: existing, alreadyConfirmed: true }
}
)

View file

@ -25,6 +25,7 @@ import { EVENT_VIEWER_URL } from './env.js'
import {
profilesByPubkey,
loadProfile,
repository,
} from './repository.js'
type DigestData = {
@ -86,7 +87,6 @@ export class Digest {
sendFromStoredEvents = async (storedEvents: { id: string; event: TrustedEvent; relay: string }[]) => {
const events = storedEvents.map(se => se.event)
const context = [...events] // For now, context == events (no reply loading)
const relayByEventId = new Map(storedEvents.map(se => [se.event.id, se.relay]))
// Load profiles for event authors
@ -101,6 +101,13 @@ export class Digest {
}
}
// Load reply/reaction context: events that tag our matched events
const eventIds = events.map(e => e.id)
const replyEvents = repository.query([
{ '#e': eventIds, kinds: [NOTE, COMMENT, REACTION] },
])
const context = [...events, ...replyEvents]
const data = { events, context, relayByEventId } as DigestData
if (data.events.length > 0) {

View file

@ -7,7 +7,9 @@ import { registerSubscription } from './worker/index.js'
process.on('unhandledRejection', (error: Error) => {
console.error('Unhandled rejection:', error.stack)
process.exit(1)
// Do not process.exit(1) — an async rejection from a library's internal
// timer (e.g. @welshman/net's batcher) would let an attacker crash the
// entire server with a single malformed request. Log and continue.
})
process.on('uncaughtException', (error: Error) => {
@ -15,6 +17,7 @@ process.on('uncaughtException', (error: Error) => {
process.exit(1)
})
migrate().then(async () => {
server.listen(PORT, () => {
console.log('Running on port', PORT)

View file

@ -0,0 +1,62 @@
<!DOCTYPE html>
<html>
<head>
<meta charset="utf-8" />
<meta name="viewport" content="width=device-width, initial-scale=1.0" />
<title>Email Already Confirmed</title>
<style>
* { box-sizing: border-box; }
body {
font-family: 'Inter', system-ui, -apple-system, sans-serif;
display: flex;
align-items: center;
justify-content: center;
min-height: 100vh;
margin: 0;
background: #f0f2f5;
color: #1e293b;
}
.container {
width: 100%;
max-width: 480px;
margin: 16px;
text-align: center;
padding: 40px 32px;
background: #ffffff;
border-radius: 12px;
border: 1px solid #e2e8f0;
}
.logo { height: 40px; width: auto; margin: 0 auto 16px; display: block; }
.brand {
font-size: 24px;
font-weight: 700;
margin: 0 0 28px;
color: {{brandAccent}};
}
.title {
font-size: 20px;
font-weight: 700;
margin: 0 0 12px;
color: #1e293b;
}
.message {
font-size: 15px;
line-height: 1.6;
color: #64748b;
margin: 0;
}
.message a { color: {{brandAccent}}; text-decoration: none; font-weight: 600; }
</style>
</head>
<body>
<div class="container">
{{#brandLogo}}<img class="logo" src="{{brandLogo}}" alt="{{brandName}}" />{{/brandLogo}}
<div class="brand">{{brandName}}</div>
<h1 class="title">Email already confirmed</h1>
<p class="message">
This email address has already been confirmed. You're all set — no further action needed.
Visit <a href="{{settingsUrl}}">{{brandName}}</a> to manage your notification settings.
</p>
</div>
</body>
</html>

View file

@ -6,7 +6,7 @@ import { render } from './templates.js'
import { confirmSubscriptionAction, unsubscribeAction, registerSubscription, ActionError } from './actions.js'
import { getSubscriptionById, insertEvent, getSubscriptionByKey, getSubscriptionByPubkey } from './database.js'
import { load } from '@welshman/net'
import { getIdFilters } from '@welshman/util'
import { getIdFilters, isRelayUrl } from '@welshman/util'
import crypto from 'crypto'
import { verifyEvent } from 'nostr-tools/pure'
@ -83,6 +83,13 @@ const verifyNip98Auth = async (req: Request): Promise<string | null> => {
export const server: express.Application = express()
// Behind a TLS-terminating reverse proxy (traefik in the ansible deploy).
// Without this, req.protocol stays "http" and verifyNip98Auth builds an
// expectedUrl of http://…, which no browser client will ever sign (clients
// sign https://…). Trust one proxy hop so req.protocol honors
// X-Forwarded-Proto and NIP-98 URL matching works.
server.set('trust proxy', 1)
// CORS middleware for browser-facing routes only.
// The browser hits /subscription with an Authorization header and Content-Type:
// application/json, which triggers a CORS preflight. Answer it and allow the
@ -256,6 +263,15 @@ addRoute('post', '/notify/:id', async (req: Request, res: Response) => {
return res.status(400).json({ error: 'id and relay are required' })
}
// Reject non-ws:// relay URLs. load() from @welshman/net throws
// Invalid relay url asynchronously inside a batcher timer, which
// escapes the route's try/catch and becomes an unhandledRejection
// that would crash the server. Validate early to avoid calling
// load() with an unsupported scheme.
if (!isRelayUrl(relay)) {
return res.status(400).json({ error: 'Invalid relay URL. Only wss:// or ws:// relays are supported.' })
}
const sub = await getSubscriptionById(req.params.id)
if (!sub) {
@ -320,7 +336,13 @@ addRoute('get', '/confirm', async (req: Request, res: Response) => {
}
try {
await confirmSubscriptionAction({ token: req.query.token })
const result = await confirmSubscriptionAction({ token: req.query.token })
if (result.alreadyConfirmed) {
return res.send(await render('pages/confirm-already.html', {
...brandingVars(),
}))
}
res.send(await render('pages/confirm-success.html', {
...brandingVars(),

View file

@ -0,0 +1,44 @@
import { describe, it, expect, beforeAll } from 'vitest'
import * as db from '../src/database.js'
const pubkey = 'reconfirm-test-' + Date.now()
const email = 'reconfirm-test-' + Date.now() + '@example.com'
let token: string
describe('Confirm link idempotency — already-confirmed token', () => {
beforeAll(async () => {
await db.migrate()
})
it('creates and confirms a subscription for the first time', async () => {
const sub = await db.insertSubscription(pubkey, email, 'daily')
expect(sub).toBeTruthy()
token = sub.key
const result = await db.confirmSubscription(token)
expect(result).toBeTruthy()
expect(result!.sub.confirmed_at).toBeTruthy()
expect(result!.alreadyConfirmed).toBe(false)
})
it('returns a distinct result (not undefined) when confirming an already-confirmed token', async () => {
// BUG: confirmSubscription used `WHERE confirmed_at IS NULL`, so
// re-confirming an already-confirmed token matched zero rows and
// the UPDATE returned nothing → parseSubscription returned undefined.
// The caller then threw ActionError('invalid or expired') and the
// user saw "Email not confirmed" — which is misleading.
//
// FIX: confirmSubscription now detects the already-confirmed case
// and returns { sub, alreadyConfirmed: true } so the handler can
// render an "already confirmed" info page instead of an error page.
const result = await db.confirmSubscription(token)
expect(result).not.toBeUndefined()
expect(result!.alreadyConfirmed).toBe(true)
expect(result!.sub.confirmed_at).toBeTruthy()
})
it('returns undefined for a nonexistent token', async () => {
const result = await db.confirmSubscription('nonexistent-token-' + Date.now())
expect(result).toBeUndefined()
})
})

View file

@ -0,0 +1,167 @@
import { describe, it, expect, vi, beforeEach } from 'vitest'
import type { TrustedEvent } from '@welshman/util'
import { spec } from '@welshman/lib'
// ── Shared state accessible from both vi.mock factories and test body ──────
const mockRepo = vi.hoisted(() => {
const events: TrustedEvent[] = []
return {
events, // shared mutable array
query: vi.fn((filters: any[]) => {
return events.filter(e => {
for (const f of filters) {
// #e filter: event must have an 'e' tag whose value matches one of the filter values
if (f['#e']) {
const eTagVals = e.tags.filter(t => t[0] === 'e').map(t => t[1])
if (!f['#e'].some((id: string) => eTagVals.includes(id))) return false
}
// kinds filter
if (f.kinds && !f.kinds.includes(e.kind)) return false
}
return true
})
}),
publish: vi.fn((event: TrustedEvent) => {
events.push(event)
return true
}),
}
})
// ── Mocks (hoisted before imports) ─────────────────────────────────────────
// ── Mock profiles map (pre-populated so waitForProfile doesn't time out) ──
const mockProfilesMap = vi.hoisted(() => new Map())
vi.mock('../src/repository.js', () => ({
profilesByPubkey: { get: () => mockProfilesMap },
loadProfile: vi.fn().mockResolvedValue(undefined),
repository: mockRepo,
}))
vi.mock('../src/util.js', () => {
const createElementMock = vi.fn().mockImplementation((tagName: string) => {
const el: any = { tagName, children: [], innerText: '' }
el.appendChild = (child: any) => { el.children.push(child) }
el.toString = () =>
`<${tagName}>${el.innerText}${el.children.map((c: any) => c.toString()).join('')}</${tagName}>`
return el
})
return {
displayDuration: vi.fn().mockReturnValue('1 hour'),
createElement: createElementMock,
}
})
// Captured digest parameters (set by the mailer mock)
let lastDigestParams: Record<string, any> | undefined
vi.mock('../src/mailer.js', () => ({
sendDigest: vi.fn((_sub: any, variables: Record<string, any>) => {
lastDigestParams = variables
}),
}))
// ── Imports (after mocks) ──────────────────────────────────────────────────
import { Digest } from '../src/digest.js'
import type { Subscription } from '../src/alert.js'
// Valid 64-char hex strings for nostr IDs and pubkeys
const PARENT_ID = 'aaa' + 'a'.repeat(61) // 64 hex chars
const REPLY_ID = 'bbb' + 'b'.repeat(61)
const REACTION_ID = 'ccc' + 'c'.repeat(61)
const PARENT_PK = 'ddd' + 'd'.repeat(61)
const REPLIER_PK = 'eee' + 'e'.repeat(61)
const REACTER_PK = 'fff' + 'f'.repeat(61)
function makeEvent(overrides: Partial<TrustedEvent>): TrustedEvent {
const eid = overrides.id || 'a'.repeat(64)
return {
id: eid,
kind: 1,
pubkey: PARENT_PK,
created_at: Math.floor(Date.now() / 1000),
tags: [],
content: 'test content',
...overrides,
id: eid,
} as TrustedEvent
}
describe('digest reply/reaction stats', () => {
let sub: Subscription
beforeEach(() => {
// Reset shared state
mockRepo.events.length = 0
lastDigestParams = undefined
sub = {
id: 'sub-1',
key: 'key-1',
pubkey: PARENT_PK,
email: 'test@example.com',
frequency: 'daily',
created_at: Math.floor(Date.now() / 1000) - 3600,
confirmed_at: Math.floor(Date.now() / 1000) - 3600,
}
})
it('should count replies and reactions loaded from repository context', async () => {
// Create a parent event
const parentEvent = makeEvent({
id: PARENT_ID,
kind: 1,
pubkey: PARENT_PK,
content: 'Hello world, this is the parent event',
created_at: Math.floor(Date.now() / 1000) - 600,
})
// Create a reply event referencing the parent via an 'e' tag
const replyEvent = makeEvent({
id: REPLY_ID,
kind: 1,
pubkey: REPLIER_PK,
content: 'This is a reply to the parent',
created_at: Math.floor(Date.now() / 1000) - 500,
tags: [['e', PARENT_ID, '', 'root']],
})
// Create a reaction event (kind 7 = REACTION)
const reactionEvent = makeEvent({
id: REACTION_ID,
kind: 7,
pubkey: REACTER_PK,
content: '+',
created_at: Math.floor(Date.now() / 1000) - 400,
tags: [['e', PARENT_ID, '', 'root']],
})
// Publish reply/reaction events to the mock repository so the fix can find them
mockRepo.publish(replyEvent)
mockRepo.publish(reactionEvent)
// Pre-populate profiles so waitForProfile resolves immediately
mockProfilesMap.set(PARENT_PK, { pubkey: PARENT_PK, name: 'parent-user', picture: '' })
mockProfilesMap.set(REPLIER_PK, { pubkey: REPLIER_PK, name: 'replier', picture: '' })
mockProfilesMap.set(REACTER_PK, { pubkey: REACTER_PK, name: 'reacter', picture: '' })
const storedEvents = [
{ id: 'se-1', event: parentEvent, relay: 'wss://relay.example.com' },
]
const digest = new Digest(sub)
await digest.sendFromStoredEvents(storedEvents)
// sendDigest should have been called with the template parameters
expect(lastDigestParams).toBeDefined()
const latest = lastDigestParams!.Latest
expect(latest.length).toBeGreaterThanOrEqual(1)
const parentEntry = latest[0]
// When context includes reply/reaction events loaded from the repository,
// Replies should be >= 1 and Reactions >= 1
expect(parentEntry.Replies).toBeGreaterThanOrEqual(1)
expect(parentEntry.Reactions).toBeGreaterThanOrEqual(1)
})
})

View file

@ -0,0 +1,50 @@
import { describe, it, expect, beforeAll } from 'vitest'
import * as db from '../src/database.js'
const pubkey = 'dup-test-' + Date.now()
const email = 'dup-test-' + Date.now() + '@example.com'
let token: string
describe('Duplicate subscription prevention — idempotent re-register after confirm', () => {
beforeAll(async () => {
await db.migrate()
})
it('registers a subscription (fresh)', async () => {
const sub = await db.insertSubscription(pubkey, email, 'daily')
expect(sub).toBeTruthy()
expect(sub!.confirmed_at).toBeFalsy()
expect(sub!.unsubscribed_at).toBeFalsy()
token = sub!.key
})
it('confirms the subscription', async () => {
const result = await db.confirmSubscription(token)
expect(result).toBeTruthy()
expect(result!.alreadyConfirmed).toBe(false)
expect(result!.sub.confirmed_at).toBeTruthy()
})
it('re-registers with the same email and frequency (idempotent PUT)', async () => {
// This simulates a second PUT from the client with identical params.
// The bug would create a second active row + send a new confirmation email.
const sub = await db.insertSubscription(pubkey, email, 'daily')
expect(sub).toBeTruthy()
// Must return the existing confirmed row, NOT a fresh unconfirmed row
expect(sub!.confirmed_at).toBeTruthy()
expect(sub!.unsubscribed_at).toBeFalsy()
// The key must remain unchanged — a new row would have a different key
expect(sub!.key).toBe(token)
})
it('has exactly one active row for this pubkey', async () => {
// getSubscriptionByPubkey filters by unsubscribed_at IS NULL and
// returns at most one row (enforced by the partial unique index).
// If a second active row exists, the index is absent or bypassed.
const active = await db.getSubscriptionByPubkey(pubkey)
expect(active).toBeTruthy()
expect(active!.confirmed_at).toBeTruthy()
expect(active!.key).toBe(token)
})
})

View file

@ -24,7 +24,7 @@ describe('Event-arrival race in digest job', () => {
expect(s).toBeTruthy()
const confirmed = await db.confirmSubscription(s.key)
expect(confirmed).toBeTruthy()
sub = confirmed
sub = confirmed!.sub
// Set last_digest_at to 60 seconds ago (the "since" value runJob would use)
since = Math.floor(Date.now() / 1000) - 60

View file

@ -0,0 +1,82 @@
// POST /notify with non-wss relay URL crashes the server (remote DoS)
//
// Bug: When POST /notify/:id receives a relay URL with a non-wss scheme
// (e.g. http://…), the handler calls `load()` from @welshman/net. Inside
// `load`, the batcher schedules an async `_execute` via setTimeout(200ms).
// When `getAdapter` throws `Invalid relay url`, the error escapes as an
// unhandledPromiseRejection because the batcher's `_execute` async function
// is called from setTimeout with no `.catch()`. The global
// `process.on('unhandledRejection')` handler in `src/index.ts` then calls
// `process.exit(1)`, killing the entire server.
//
// Fix applied (2 of 3 fixes):
// 1. Validate relay scheme before calling load() — the route handler now
// checks isRelayUrl() and returns 400 for non-ws:// schemes.
// 2. Don't process.exit(1) on unhandledRejection — log and continue
// (defence in depth for any other async edge-case).
import { describe, it, expect, beforeAll, afterAll } from 'vitest'
import * as db from '../src/database.js'
import { server } from '../src/server.js'
import { createServer, type Server } from 'http'
// Partially mock @welshman/net so that `load()` returns an empty array,
// preventing real relay connections during the test, while preserving all
// other exports from the library.
vi.mock('@welshman/net', async (importOriginal) => {
const actual = await importOriginal()
return {
...(actual as Record<string, unknown>),
load: vi.fn().mockResolvedValue([]),
}
})
describe('notify_non_wss_relay', () => {
let httpServer: Server
let baseUrl: string
let subId: string
beforeAll(async () => {
await db.migrate()
// Create and confirm a subscription we can use for the notify call
const pubkey = 'nws-test-pk-' + Date.now()
const email = 'nws-test-' + Date.now() + '@example.com'
const sub = await db.insertSubscription(pubkey, email, 'daily')
const confirmed = await db.confirmSubscription(sub.key)
subId = confirmed.id
// Start the express server on a random available port
await new Promise<void>((resolve) => {
httpServer = createServer(server)
httpServer.listen(0, () => {
const addr = httpServer.address()
if (addr && typeof addr === 'object') {
baseUrl = `http://localhost:${addr.port}`
}
resolve()
})
})
})
afterAll(async () => {
httpServer?.close()
})
it('rejects non-wss relay URL with 400 instead of crashing the server', async () => {
const res = await fetch(`${baseUrl}/notify/${subId}`, {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify({
id: 'nonexistent-' + Date.now(),
relay: 'http://127.0.0.1:9',
}),
})
expect(res.status).toBe(400)
const body = await res.json()
expect(body).toHaveProperty('error')
expect(body.error).toMatch(/Invalid relay/)
})
})

View file

@ -37,7 +37,7 @@ describe('notify_response_shape', () => {
const email = 'shape-test-' + Date.now() + '@example.com'
const sub = await db.insertSubscription(pubkey, email, 'daily')
const confirmed = await db.confirmSubscription(sub.key)
subId = confirmed.id
subId = confirmed!.sub.id
// Start the express server on a random available port
await new Promise<void>((resolve) => {

View file

@ -20,7 +20,7 @@ describe('Frequency change reschedules cron job', () => {
const confirmed = await db.confirmSubscription(sub.key)
expect(confirmed).toBeTruthy()
sub = confirmed
sub = confirmed!.sub
})
it('registers cron job with daily frequency', () => {