/**
* logship-mail-archive — small read-only HTTP API in front of the mailpiler archive.
*
* Runs ON the piler host (192.168.9.53, /opt/logship-mail-archive, systemd unit
* `logship-mail-archive`, port 8097) and is called by the ERP (Nuxt server,
* server/utils/mail/mailArchive.ts) to show a contact's e-mail history.
*
* GET /health
* GET /scope → { uid, addresses } the search is limited to
* GET /search?email=
[&limit=300] → { rows: [...] , ms }
* GET /message?id=&email= → { message: { html, text, … } }
* GET /attachment?id=&email=&index= → the attachment (binary)
*
* Every request needs `Authorization: Bearer `.
* SCOPE: the archive is shared by several companies. The service only ever returns
* mails in which one of the addresses of ONE piler account (PILER_UID, table
* `email`; uid 5 = "logou" / LogYou) is sender or recipient — the same visibility
* that account has in the piler web GUI. New addresses added to the account in
* piler are picked up within 5 minutes.
* /message and /attachment require the contact's address and refuse a message
* that does not involve it — the API cannot be used to read arbitrary mail by id.
*
* How it reads piler (nothing is written, piler itself is untouched):
* - address lookup: MariaDB `metadata.from` / `rcpt.to` (both indexed, ~5 ms on
* 600k mails). Credentials come from /usr/local/etc/piler/piler.conf.
* - message content: `pilerget ` (decrypts + decompresses the stored
* message, ~0.25 s for 30 MB), parsed with mailparser. Exchange journal
* reports are unwrapped to the original message.
* - Manticore (127.0.0.1:9306, index `piler1`) is NOT needed for address
* lookups; it is the place to go for a later full-text search endpoint
* (`MATCH('@(subject,body) …')`, addresses are indexed with @ and . → X).
*/
import http from 'node:http'
import { spawn } from 'node:child_process'
import { readFileSync } from 'node:fs'
import { timingSafeEqual } from 'node:crypto'
import mysql from 'mysql2/promise'
import { simpleParser } from 'mailparser'
const PORT = Number(process.env.PORT || 8097)
const HOST = process.env.HOST || '0.0.0.0'
const TOKEN = String(process.env.MAIL_ARCHIVE_TOKEN || '')
const PILER_CONF = process.env.PILER_CONF || '/usr/local/etc/piler/piler.conf'
const PILERGET = process.env.PILERGET || '/usr/local/bin/pilerget'
const PILER_UID = Number(process.env.PILER_UID)
const MAX_MESSAGE_BYTES = 80 * 1024 * 1024
const CACHE_ENTRIES = 6
const CACHE_TTL_MS = 10 * 60 * 1000
if (!Number.isInteger(PILER_UID) || PILER_UID < 2) { console.error('PILER_UID missing (piler account whose addresses limit the search)'); process.exit(1) }
if (TOKEN.length < 24) { console.error('MAIL_ARCHIVE_TOKEN missing or too short'); process.exit(1) }
const conf = Object.fromEntries(readFileSync(PILER_CONF, 'utf8').split('\n')
.filter((l) => l && !l.startsWith(';') && !l.startsWith('#') && l.includes('='))
.map((l) => [l.slice(0, l.indexOf('=')).trim(), l.slice(l.indexOf('=') + 1).trim()]))
const pool = mysql.createPool({
socketPath: conf.mysqlsocket || '/var/run/mysqld/mysqld.sock',
user: conf.mysqluser || 'piler',
password: conf.mysqlpwd || '',
database: conf.mysqldb || 'piler',
charset: 'utf8mb4',
connectionLimit: 4
})
const EMAIL_RE = /^[a-z0-9._%+\-]+@[a-z0-9.\-]+\.[a-z0-9\-]{2,}$/
const cleanEmail = (v) => { const e = String(v || '').trim().toLowerCase(); return EMAIL_RE.test(e) && e.length <= 128 ? e : '' }
const str = (v) => (Buffer.isBuffer(v) ? v.toString('utf8') : String(v ?? ''))
const send = (res, status, body) => {
const data = JSON.stringify(body)
res.writeHead(status, { 'Content-Type': 'application/json; charset=utf-8', 'Content-Length': Buffer.byteLength(data), 'Cache-Control': 'no-store' })
res.end(data)
}
const authorized = (req) => {
const given = Buffer.from(String(req.headers.authorization || '').replace(/^Bearer\s+/i, ''))
const want = Buffer.from(TOKEN)
return given.length === want.length && timingSafeEqual(given, want)
}
/** Addresses of the piler account the service is limited to (cached 5 min). */
let scopeCache = { at: 0, list: [] }
const scope = async () => {
if (Date.now() - scopeCache.at < 5 * 60 * 1000 && scopeCache.list.length) return scopeCache.list
const [rows] = await pool.query('SELECT email FROM email WHERE uid = ?', [PILER_UID])
scopeCache = { at: Date.now(), list: rows.map((r) => str(r.email).toLowerCase()).filter(Boolean) }
return scopeCache.list
}
// "one of the account's addresses is sender or recipient of m"
const IN_SCOPE = '(m.`from` IN (?) OR EXISTS (SELECT 1 FROM rcpt s WHERE s.id = m.id AND s.`to` IN (?)))'
/** Every archived mail between `email` and the account's addresses, newest first. */
const search = async (email, limit) => {
const own = await scope()
if (!own.length) return []
const [rows] = await pool.query(
'SELECT m.id, m.`from`, m.subject, m.sent, m.arrived, m.size, m.message_id ' +
'FROM metadata m JOIN (SELECT id FROM metadata WHERE `from` = ? UNION SELECT id FROM rcpt WHERE `to` = ?) x ON x.id = m.id ' +
'WHERE m.deleted = 0 AND ' + IN_SCOPE + ' ORDER BY m.sent DESC LIMIT ?', [email, email, own, own, limit])
if (!rows.length) return []
const [rcpts] = await pool.query('SELECT id, `to` FROM rcpt WHERE id IN (?)', [rows.map((r) => r.id)])
const to = new Map()
for (const r of rcpts) { const k = String(r.id); if (!to.has(k)) to.set(k, []); to.get(k).push(str(r.to)) }
return rows.map((r) => {
const from = str(r.from).toLowerCase()
const ts = Number(r.sent) || Number(r.arrived) || 0
return {
id: Number(r.id),
messageId: str(r.message_id),
date: ts ? new Date(ts * 1000).toISOString() : null,
subject: str(r.subject),
from: { name: '', address: from },
to: (to.get(String(r.id)) || []).map((a) => ({ name: '', address: a })),
direction: from === email ? 'in' : 'out',
size: Number(r.size) || 0
}
})
}
/** piler_id of a message, only if it involves `email` AND the account's addresses. */
const pilerIdFor = async (id, email) => {
const own = await scope()
if (!own.length) return ''
const [rows] = await pool.query(
'SELECT m.piler_id FROM metadata m WHERE m.id = ? AND m.deleted = 0 AND (m.`from` = ? OR EXISTS (SELECT 1 FROM rcpt r WHERE r.id = m.id AND r.`to` = ?)) AND ' + IN_SCOPE,
[id, email, email, own, own])
const pid = rows[0] ? str(rows[0].piler_id) : ''
return /^[0-9a-f]{36}$/i.test(pid) ? pid : ''
}
const pilerget = (pilerId) => new Promise((resolve, reject) => {
const p = spawn(PILERGET, [pilerId], { stdio: ['ignore', 'pipe', 'pipe'] })
const chunks = []
let size = 0
let err = ''
p.stdout.on('data', (c) => { size += c.length; if (size > MAX_MESSAGE_BYTES) { p.kill(); reject(new Error('Message too large')) } else chunks.push(c) })
p.stderr.on('data', (c) => { err += c })
p.on('error', reject)
p.on('close', (code) => (size ? resolve(Buffer.concat(chunks)) : reject(new Error(`pilerget failed (${code}) ${err.slice(0, 200)}`))))
})
const cache = new Map() // id → { at, mail } — parsed mails incl. attachment buffers
const parsedMail = async (id, pilerId) => {
const hit = cache.get(id)
if (hit && Date.now() - hit.at < CACHE_TTL_MS) return hit.mail
let mail = await simpleParser(await pilerget(pilerId))
// Exchange journaling wraps the original message in a journal report.
const inner = (mail.attachments || []).find((a) => a.contentType === 'message/rfc822')
if (inner && (mail.headers.has('x-ms-journal-report') || !mail.html)) mail = await simpleParser(inner.content)
cache.set(id, { at: Date.now(), mail })
while (cache.size > CACHE_ENTRIES) cache.delete(cache.keys().next().value)
return mail
}
const addresses = (v) => (Array.isArray(v) ? v : v ? [v] : []).flatMap((x) => x?.value || [])
.map((a) => ({ name: String(a?.name || ''), address: String(a?.address || '') }))
const realAttachments = (mail) => (mail.attachments || []).filter((a) => !(a.related && a.cid))
const server = http.createServer(async (req, res) => {
const started = Date.now()
const url = new URL(req.url, 'http://x')
try {
if (req.method !== 'GET') return send(res, 405, { message: 'Method not allowed' })
if (url.pathname === '/health') return send(res, 200, { ok: true })
if (url.pathname === '/scope') { if (!authorized(req)) return send(res, 401, { message: 'Unauthorized' }); return send(res, 200, { uid: PILER_UID, addresses: await scope() }) }
if (!authorized(req)) return send(res, 401, { message: 'Unauthorized' })
const email = cleanEmail(url.searchParams.get('email'))
if (!email) return send(res, 400, { message: 'Invalid e-mail address' })
if (url.pathname === '/search') {
const limit = Math.min(1000, Math.max(1, Number(url.searchParams.get('limit')) || 300))
const rows = await search(email, limit)
return send(res, 200, { rows, truncated: rows.length >= limit, ms: Date.now() - started })
}
if (url.pathname === '/message' || url.pathname === '/attachment') {
const id = Number(url.searchParams.get('id')) || 0
const pilerId = id ? await pilerIdFor(id, email) : ''
if (!pilerId) return send(res, 404, { message: 'Message not found' })
const mail = await parsedMail(id, pilerId)
const atts = realAttachments(mail)
if (url.pathname === '/attachment') {
const a = atts[Number(url.searchParams.get('index'))]
if (!a) return send(res, 404, { message: 'Attachment not found' })
const name = String(a.filename || 'attachment').replace(/[\r\n"]/g, '')
res.writeHead(200, {
'Content-Type': a.contentType || 'application/octet-stream',
'Content-Length': a.content.length,
'Content-Disposition': `attachment; filename*=UTF-8''${encodeURIComponent(name)}`,
'Cache-Control': 'no-store'
})
return res.end(a.content)
}
return send(res, 200, {
ms: Date.now() - started,
message: {
id,
messageId: String(mail.messageId || ''),
date: mail.date ? new Date(mail.date).toISOString() : null,
subject: String(mail.subject || ''),
from: addresses(mail.from),
to: addresses(mail.to),
cc: addresses(mail.cc),
html: typeof mail.html === 'string' ? mail.html : '', // cid: images already inlined as data URIs
text: String(mail.text || ''),
attachments: atts.map((a, index) => ({ index, filename: String(a.filename || 'attachment'), contentType: String(a.contentType || ''), size: Number(a.size || 0) }))
}
})
}
return send(res, 404, { message: 'Not found' })
} catch (err) {
console.error(new Date().toISOString(), url.pathname, err?.message || err)
if (!res.headersSent) send(res, 500, { message: 'Archive error' })
}
})
server.listen(PORT, HOST, () => console.log(`logship-mail-archive listening on ${HOST}:${PORT}`))