Compare commits
4
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
 jochen
|
be53f5167a | ||
|
jochen
|
2130f99612 | ||
|
jochen
|
ed5690cee4 | ||
|
jochen
|
70c1110c53 |
@@ -14,7 +14,7 @@ WORKDIR /src
|
||||
COPY go.mod go.sum ./
|
||||
RUN go mod download
|
||||
COPY . .
|
||||
RUN CGO_ENABLED=0 GOOS=linux go build -trimpath -buildvcs=false -ldflags=-buildid= -o /mesh-builder ./cmd/mesh-builder
|
||||
RUN CGO_ENABLED=0 GOOS=linux go build -trimpath -o /mesh-builder ./cmd/mesh-builder
|
||||
|
||||
# Unlike mesh-controller's own FROM scratch (ADR 0006: nothing to audit but one binary), the build
|
||||
# machine's whole job is shelling out to git and docker — it needs a real userland to do that in,
|
||||
|
||||
@@ -914,11 +914,6 @@ export class GiteaAdmin {
|
||||
return { status: res.status, body };
|
||||
}
|
||||
|
||||
/** One admin-API call with a JSON body, answering the status and body (the system webhook, novox/hq issue 729). */
|
||||
call(method: string, path: string, body?: unknown): Promise<AdminResponse> {
|
||||
return this.request(path, { method, ...(body === undefined ? {} : { body: JSON.stringify(body) }) });
|
||||
}
|
||||
|
||||
/** Fail with the forge's own message when a status the caller did not expect comes back. */
|
||||
private static fail(path: string, res: AdminResponse): never {
|
||||
const detail = typeof res.body === "string" ? res.body : JSON.stringify(res.body);
|
||||
|
||||
+64
-152
@@ -40,11 +40,10 @@ import { emit, on } from "@novox/mesh-sdk/events";
|
||||
import { registerModuleTools } from "@novox/mesh-sdk/tools";
|
||||
import { commitStatusTool } from "./delivery.js";
|
||||
import { mergeTool } from "./merge.js";
|
||||
import { GiteaAdmin, GiteaClient, movedSince } from "./client.js";
|
||||
import { GiteaClient, movedSince } from "./client.js";
|
||||
import { CHECK_CONTEXT, REPO_CHECK_CONTEXT, commentFor, headsToAnnounce, statusesFor, type Announced, type Checked,
|
||||
type RepoCheckFacts, reposToLook } from "./pulls.js";
|
||||
import { GateWriter, clonedAs, type BuildStarted } from "./gate.js";
|
||||
import { HOOK_PATH, HOOK_PORT, Pace, claimHead, ensureSystemHook, keptSecret, receiver, type ForgeEvent } from "./webhook.js";
|
||||
import { announcementOf, keepTook, lateVerdictComment, outOfDateWords, sameCommit, tookFile, weighTrunks } from "./trunk.js";
|
||||
|
||||
// Without a way to a token — configured, or mintable with the admin account (token.ts) — there is
|
||||
@@ -146,9 +145,57 @@ async function pollMerged(client: GiteaClient): Promise<void> {
|
||||
// at once.
|
||||
const fresh = !!pull.merged_at && !!since && pull.merged_at > since;
|
||||
if (primedMerges && fresh) {
|
||||
if (await announceMerge(client, repo, pull)) pace.miss(`the merge of ${repo.full_name}#${pull.number}`);
|
||||
// What it changed, asked for only now: a module is rebuilt because a file inside its own
|
||||
// directory moved, and without this every module built from a repository is rebuilt for a
|
||||
// change to any of them (novox/hq 04-ISSUES/131).
|
||||
const changed = await client.listPullFiles(repo.owner, repo.name, pull.number);
|
||||
// Which of the directories they are in hold a module at the merge commit (novox/hq issue 278): a
|
||||
// change inside one is that module's, held or not, and only a file in none is shared code. Not
|
||||
// said when the list is cut or the forge could not be asked; the mesh then keeps its old rule.
|
||||
let moduleDirs: string[] | null = null;
|
||||
if (!changed.truncated) {
|
||||
try {
|
||||
moduleDirs = await client.moduleDirsAt(repo.owner, repo.name, pull.merge_commit_sha, changed.paths);
|
||||
} catch (err) {
|
||||
console.error(`[gitea] ${repo.full_name}#${pull.number}: which directories hold a module could not be read, ` +
|
||||
`so the mesh reads its files by the old rule — ${err instanceof Error ? err.message : String(err)}`);
|
||||
}
|
||||
}
|
||||
// The head it merged, and what its head was checked as (novox/hq ADR 0239): the delivery it was, made
|
||||
// from the forge's word when its owner never heard the head.
|
||||
let headChecks: Record<string, string> | null = null;
|
||||
if (pull.head_sha) {
|
||||
try {
|
||||
headChecks = await client.commitStatuses(repo.owner, repo.name, pull.head_sha);
|
||||
} catch (err) {
|
||||
console.error(`[gitea] ${repo.full_name}#${pull.number}: its head's statuses could not be read — ${err instanceof Error ? err.message : err}`);
|
||||
}
|
||||
}
|
||||
await emit("pull.merged", {
|
||||
owner: repo.owner,
|
||||
repo: repo.name,
|
||||
number: pull.number,
|
||||
title: pull.title,
|
||||
body: pull.body,
|
||||
head_sha: pull.head_sha,
|
||||
...(headChecks ? { head_checks: headChecks } : {}),
|
||||
head: pull.head,
|
||||
base: pull.base,
|
||||
merge_commit_sha: pull.merge_commit_sha,
|
||||
merged_at: pull.merged_at,
|
||||
clone_url: repo.clone_url,
|
||||
html_url: pull.html_url,
|
||||
paths: changed.paths,
|
||||
paths_truncated: changed.truncated,
|
||||
// Which of them the merge deleted (novox/hq ADR 0236): a module whose manifest went is forgotten,
|
||||
// not built.
|
||||
removed: changed.removed,
|
||||
...(moduleDirs ? { module_dirs: moduleDirs, module_dirs_said: true } : {}),
|
||||
});
|
||||
// Said, because a trigger that fires silently is indistinguishable from one that did not
|
||||
// fire (novox/hq 04-ISSUES/131) — this line is how an operator knows the mesh was told.
|
||||
console.log(`[gitea] announced merge ${repo.full_name}#${pull.number} (${pull.merge_commit_sha.slice(0, 8)}) into ${pull.base}`);
|
||||
}
|
||||
else announced.add(pull.merge_commit_sha);
|
||||
announced.add(pull.merge_commit_sha);
|
||||
changed = true;
|
||||
}
|
||||
@@ -159,73 +206,6 @@ async function pollMerged(client: GiteaClient): Promise<void> {
|
||||
lastLook = began;
|
||||
}
|
||||
|
||||
type RepoOf = { owner: string; name: string; full_name: string; clone_url?: string };
|
||||
type PullOf = Awaited<ReturnType<GiteaClient["listPullRequests"]>>[number];
|
||||
|
||||
/** **A merge, announced once, whichever path saw it first** (novox/hq issue 729): the poll or the forge's event. */
|
||||
async function announceMerge(client: GiteaClient, repo: RepoOf, pull: PullOf): Promise<boolean> {
|
||||
if (!pull.merged || !pull.merge_commit_sha || announced.has(pull.merge_commit_sha)) return false;
|
||||
// Recorded before the first wait, so the other path finds it taken while this one is still asking the forge.
|
||||
announced.add(pull.merge_commit_sha);
|
||||
try {
|
||||
// What it changed, asked for only now: a module is rebuilt because a file inside its own
|
||||
// directory moved, and without this every module built from a repository is rebuilt for a
|
||||
// change to any of them (novox/hq 04-ISSUES/131).
|
||||
const changed = await client.listPullFiles(repo.owner, repo.name, pull.number);
|
||||
// Which of the directories they are in hold a module at the merge commit (novox/hq issue 278): a
|
||||
// change inside one is that module's, held or not, and only a file in none is shared code. Not
|
||||
// said when the list is cut or the forge could not be asked; the mesh then keeps its old rule.
|
||||
let moduleDirs: string[] | null = null;
|
||||
if (!changed.truncated) {
|
||||
try {
|
||||
moduleDirs = await client.moduleDirsAt(repo.owner, repo.name, pull.merge_commit_sha, changed.paths);
|
||||
} catch (err) {
|
||||
console.error(`[gitea] ${repo.full_name}#${pull.number}: which directories hold a module could not be read, ` +
|
||||
`so the mesh reads its files by the old rule — ${err instanceof Error ? err.message : String(err)}`);
|
||||
}
|
||||
}
|
||||
// The head it merged, and what its head was checked as (novox/hq ADR 0239): the delivery it was, made
|
||||
// from the forge's word when its owner never heard the head.
|
||||
let headChecks: Record<string, string> | null = null;
|
||||
if (pull.head_sha) {
|
||||
try {
|
||||
headChecks = await client.commitStatuses(repo.owner, repo.name, pull.head_sha);
|
||||
} catch (err) {
|
||||
console.error(`[gitea] ${repo.full_name}#${pull.number}: its head's statuses could not be read — ${err instanceof Error ? err.message : err}`);
|
||||
}
|
||||
}
|
||||
await emit("pull.merged", {
|
||||
owner: repo.owner,
|
||||
repo: repo.name,
|
||||
number: pull.number,
|
||||
title: pull.title,
|
||||
body: pull.body,
|
||||
head_sha: pull.head_sha,
|
||||
...(headChecks ? { head_checks: headChecks } : {}),
|
||||
head: pull.head,
|
||||
base: pull.base,
|
||||
merge_commit_sha: pull.merge_commit_sha,
|
||||
merged_at: pull.merged_at,
|
||||
clone_url: repo.clone_url,
|
||||
html_url: pull.html_url,
|
||||
paths: changed.paths,
|
||||
paths_truncated: changed.truncated,
|
||||
// Which of them the merge deleted (novox/hq ADR 0236): a module whose manifest went is forgotten,
|
||||
// not built.
|
||||
removed: changed.removed,
|
||||
...(moduleDirs ? { module_dirs: moduleDirs, module_dirs_said: true } : {}),
|
||||
});
|
||||
// Said, because a trigger that fires silently is indistinguishable from one that did not
|
||||
// fire (novox/hq 04-ISSUES/131) — this line is how an operator knows the mesh was told.
|
||||
console.log(`[gitea] announced merge ${repo.full_name}#${pull.number} (${pull.merge_commit_sha.slice(0, 8)}) into ${pull.base}`);
|
||||
} catch (err) {
|
||||
announced.delete(pull.merge_commit_sha);
|
||||
throw err;
|
||||
}
|
||||
keepAnnounced();
|
||||
return true;
|
||||
}
|
||||
|
||||
// **Every new head of an open pull request is announced, once** (novox/hq to-be 45 §9): the mesh checks it
|
||||
// against every machine of its facts before it merges. Kept beside the merges' record, so a restart
|
||||
// announces nothing twice; the first look on a machine with no record announces only what moved in the
|
||||
@@ -248,31 +228,6 @@ function keepHeads(): void {
|
||||
writeFileSync(tmp, JSON.stringify({ heads }));
|
||||
renameSync(tmp, pullsRecord);
|
||||
}
|
||||
/** **A head, announced once, whichever path saw it first** (novox/hq issue 729): the poll or the forge's event. */
|
||||
async function announceHead(client: GiteaClient, repo: RepoOf, pull: OpenOf): Promise<boolean> {
|
||||
const key = `${repo.full_name}#${pull.number}`;
|
||||
const sha = String(pull.head_sha ?? "");
|
||||
const before = heads[key];
|
||||
// Claimed before the first wait, so the other path finds it taken while this one is still announcing.
|
||||
if (!claimHead(heads, key, sha)) return false;
|
||||
try {
|
||||
// The announcement the merge tool makes too when it asks a check (trunk.ts, novox/hq ADR 0323 §5).
|
||||
await emit("pull.updated", await announcementOf(client, repo.owner, repo.name, repo.clone_url ?? "", pull));
|
||||
} catch (err) {
|
||||
if (before === undefined) delete heads[key];
|
||||
else heads[key] = before;
|
||||
throw err;
|
||||
}
|
||||
keepHeads();
|
||||
// Said, so an operator knows the mesh was asked to check it.
|
||||
console.log(`[gitea] announced ${key} at ${sha.slice(0, 8)} to be checked before it merges`);
|
||||
// Pending until the verdict comes, so the pull request says a check is running rather than nothing — and
|
||||
// where it is: in the build queue from now, until a build machine takes it (novox/hq issue 446).
|
||||
await gates.pending(repo.owner, repo.name, sha);
|
||||
return true;
|
||||
}
|
||||
type OpenOf = Awaited<ReturnType<GiteaClient["listOpenPullsSince"]>>[number];
|
||||
|
||||
let lastPullLook = "";
|
||||
// **A sweep now and then looks at every repository** (novox/hq issue 347), whatever moved: the floor is a
|
||||
// heuristic, and a heuristic that misses is caught by the next sweep rather than never.
|
||||
@@ -304,8 +259,15 @@ async function pollPulls(client: GiteaClient): Promise<void> {
|
||||
const key = `${repo.full_name}#${pull.number}`;
|
||||
const fresh = primedPulls || (!!pull.updated_at && pull.updated_at > dayAgo);
|
||||
if (fresh) {
|
||||
if (await announceHead(client, repo, pull)) pace.miss(`${key} at ${String(pull.head_sha).slice(0, 8)}`);
|
||||
} else heads[key] = String(pull.head_sha);
|
||||
// The announcement the merge tool makes too when it asks a check (trunk.ts, novox/hq ADR 0323 §5).
|
||||
await emit("pull.updated", await announcementOf(client, repo.owner, repo.name, repo.clone_url, pull));
|
||||
// Said, so an operator knows the mesh was asked to check it.
|
||||
console.log(`[gitea] announced ${key} at ${String(pull.head_sha).slice(0, 8)} to be checked before it merges`);
|
||||
// Pending until the verdict comes, so the pull request says a check is running rather than nothing — and
|
||||
// where it is: in the build queue from now, until a build machine takes it (novox/hq issue 446).
|
||||
await gates.pending(repo.owner, repo.name, String(pull.head_sha));
|
||||
}
|
||||
heads[key] = String(pull.head_sha);
|
||||
changed = true;
|
||||
}
|
||||
if (await trunksMoved(client, repo.owner, repo.name)) changedTrunks = true;
|
||||
@@ -451,59 +413,12 @@ async function buildStarted(event: { body: unknown }): Promise<void> {
|
||||
}
|
||||
}
|
||||
|
||||
/** How fast the polls look: 30 s until the forge's events are heard, five minutes while they are (webhook.ts). */
|
||||
const pace = new Pace();
|
||||
|
||||
/** What the forge's event asks of the announcer at once (issue 729): a new head is announced, and a merge. */
|
||||
async function actOn(client: GiteaClient, event: ForgeEvent): Promise<void> {
|
||||
const b = event.body as { owner?: string; repo?: string; number?: number };
|
||||
if (!b.owner || !b.repo || !b.number) return;
|
||||
const head = ["forge.pull-request.opened", "forge.pull-request.synchronized", "forge.pull-request.reopened"].includes(event.name);
|
||||
const merged = event.name === "forge.pull-request.merged";
|
||||
if (!head && !merged) return;
|
||||
const [repo, pull] = await Promise.all([client.getRepo(b.owner, b.repo), client.getPullRequest(b.owner, b.repo, b.number)]);
|
||||
if (head && pull.state === "open") await announceHead(client, repo, pull);
|
||||
if (merged) await announceMerge(client, repo, pull);
|
||||
}
|
||||
|
||||
/**
|
||||
* **The receiver, and the forge's one system webhook pointing at it** (issue 729). The secret is minted and kept beside
|
||||
* the token; the webhook is made or corrected with the admin account (ADR 0266: the tools' token holds no admin scope).
|
||||
* Answers whether events are heard; a forge that refuses is said, and the polls keep their pace.
|
||||
*/
|
||||
async function listenForForge(client: GiteaClient): Promise<boolean> {
|
||||
const stateDir = process.env.MESH_GITEA_STATE_DIR;
|
||||
const url = process.env.MESH_GITEA_HOOK_URL;
|
||||
if (!stateDir || !url) {
|
||||
console.log("[gitea] no webhook receiver: MESH_GITEA_STATE_DIR or MESH_GITEA_HOOK_URL unset, so the polls alone watch the forge");
|
||||
return false;
|
||||
}
|
||||
try {
|
||||
const secret = keptSecret(join(stateDir, "webhook.secret"));
|
||||
const port = Number(process.env.MESH_GITEA_HOOK_PORT ?? HOOK_PORT);
|
||||
const server = receiver({ secret, port, heard: () => pace.heard(),
|
||||
publish: (event) => emit(event.name, event.body),
|
||||
act: (event) => actOn(client, event) });
|
||||
await new Promise<void>((resolve, reject) => {
|
||||
server.once("error", reject);
|
||||
server.listen(port, process.env.MESH_GITEA_HOOK_HOST ?? "0.0.0.0", () => resolve());
|
||||
});
|
||||
const did = await ensureSystemHook(GiteaAdmin.fromEnv(), url.replace(/\/+$/, "") + HOOK_PATH, secret);
|
||||
console.log(`[gitea] listening for the forge's events on port ${port}; its system webhook ${did}, every event, signed`);
|
||||
return true;
|
||||
} catch (err) {
|
||||
console.error(`[gitea] the forge's events are not heard, so the polls keep their pace — ${err instanceof Error ? err.message : err}`);
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
if (gitea) {
|
||||
const client = gitea;
|
||||
// A poll that fails says so once, not once a minute: the same reason repeating (the forge not up
|
||||
// yet, the admin account refused on a restored forge) is one fact, and a recovery is worth a line.
|
||||
let failing: string | null = null;
|
||||
const tick = (fn: () => Promise<void>, every: number | (() => number)): void => {
|
||||
const everyMs = (): number => (typeof every === "number" ? every : every());
|
||||
const tick = (fn: () => Promise<void>, everyMs: number): void => {
|
||||
// **One pass at a time** (novox/hq issue 250): the next pass is scheduled when this one has ended, so a
|
||||
// pass that outlasts its interval delays the next instead of running beside it — two passes at once
|
||||
// could each announce the same merge before either recorded it.
|
||||
@@ -518,15 +433,12 @@ if (gitea) {
|
||||
if (why !== failing) console.error(`[gitea] not watching until this clears — ${why}`);
|
||||
failing = why;
|
||||
})
|
||||
.finally(() => setTimeout(run, everyMs()));
|
||||
.finally(() => setTimeout(run, everyMs));
|
||||
run();
|
||||
};
|
||||
// **The forge's events first, the polls as the safety net** (novox/hq issue 729): while verified deliveries arrive the
|
||||
// polls slow to every five minutes and announce only what no event said (Pace); otherwise they keep 30 s.
|
||||
await listenForForge(client);
|
||||
tick(() => pollRepos(client), 60_000);
|
||||
tick(() => pollMerged(client), () => pace.every());
|
||||
tick(() => pollPulls(client), () => pace.every());
|
||||
tick(() => pollMerged(client), 30_000);
|
||||
tick(() => pollPulls(client), 30_000);
|
||||
await on("mesh-controller.checked", (event) => {
|
||||
const c = (event.body ?? {}) as Checked;
|
||||
// A head's own verdict takes its turn after every write of that head's gate already asked for (gate.ts).
|
||||
|
||||
@@ -42,47 +42,7 @@
|
||||
"issue.opened",
|
||||
"pull.merged",
|
||||
"pull.updated",
|
||||
"pull.closed",
|
||||
"forge.push",
|
||||
"forge.create",
|
||||
"forge.delete",
|
||||
"forge.fork",
|
||||
"forge.status",
|
||||
"forge.pull-request.opened",
|
||||
"forge.pull-request.reopened",
|
||||
"forge.pull-request.synchronized",
|
||||
"forge.pull-request.edited",
|
||||
"forge.pull-request.closed",
|
||||
"forge.pull-request.merged",
|
||||
"forge.pull-request.labeled",
|
||||
"forge.pull-request.unlabeled",
|
||||
"forge.pull-request.assigned",
|
||||
"forge.pull-request.unassigned",
|
||||
"forge.pull-request.reviewed",
|
||||
"forge.pull-request.review-requested",
|
||||
"forge.pull-request.commented",
|
||||
"forge.pull-request.changed",
|
||||
"forge.issue-comment.created",
|
||||
"forge.issue-comment.edited",
|
||||
"forge.issue-comment.deleted",
|
||||
"forge.issues.opened",
|
||||
"forge.issues.reopened",
|
||||
"forge.issues.closed",
|
||||
"forge.issues.edited",
|
||||
"forge.issues.labeled",
|
||||
"forge.issues.unlabeled",
|
||||
"forge.issues.assigned",
|
||||
"forge.issues.unassigned",
|
||||
"forge.issues.changed",
|
||||
"forge.release.published",
|
||||
"forge.release.updated",
|
||||
"forge.release.deleted",
|
||||
"forge.release.changed",
|
||||
"forge.repository.created",
|
||||
"forge.repository.deleted",
|
||||
"forge.repository.changed",
|
||||
"forge.wiki.changed",
|
||||
"forge.other"
|
||||
"pull.closed"
|
||||
],
|
||||
"consumes": [
|
||||
"mesh-controller.checked",
|
||||
@@ -101,14 +61,7 @@
|
||||
"port": 22,
|
||||
"protocol": "tcp",
|
||||
"from": "mesh",
|
||||
"why": "git over ssh, gitea's own unmodified sshd. Published on the machine's own side at 222, the mesh's fixed public convention — not 22, which the machine's own daemon holds and a module does not take"
|
||||
},
|
||||
{
|
||||
"name": "forge-events",
|
||||
"port": 3047,
|
||||
"protocol": "tcp",
|
||||
"from": "machine",
|
||||
"why": "the forge's system webhook, every event signed with a secret the module keeps, from the forge's container on the docker bridge to the module's receiver at the bridge's gateway 172.17.0.1 (novox/hq issue 729)"
|
||||
"why": "git over ssh, gitea's own unmodified sshd. Published on the machine's own side at 222, the mesh's fixed public convention \u2014 not 22, which the machine's own daemon holds and a module does not take"
|
||||
}
|
||||
],
|
||||
"serves": {
|
||||
@@ -200,7 +153,7 @@
|
||||
"type": "file",
|
||||
"path": "${dir:state}/server.env",
|
||||
"mode": "0600",
|
||||
"content": "GITEA__security__INTERNAL_TOKEN=${secret:internal-token}\nGITEA__database__DB_TYPE=postgres\nGITEA__database__HOST=${bound:postgres-database:at}:${bound:postgres-database:port}\nGITEA__database__NAME=${bound:postgres-database:as}\nGITEA__database__USER=${bound:postgres-database:as}\nGITEA__database__PASSWD=${secret:postgres-database}\nGITEA__webhook__ALLOWED_HOST_LIST=172.17.0.1\n"
|
||||
"content": "GITEA__security__INTERNAL_TOKEN=${secret:internal-token}\nGITEA__database__DB_TYPE=postgres\nGITEA__database__HOST=${bound:postgres-database:at}:${bound:postgres-database:port}\nGITEA__database__NAME=${bound:postgres-database:as}\nGITEA__database__USER=${bound:postgres-database:as}\nGITEA__database__PASSWD=${secret:postgres-database}\n"
|
||||
},
|
||||
{
|
||||
"id": "data",
|
||||
@@ -310,10 +263,7 @@
|
||||
"MESH_GITEA_ADMIN_PASSWORD_FILE": "${dir:state}/admin.secret",
|
||||
"MESH_GITEA_STATE_DIR": "${dir:runtime-state}",
|
||||
"MESH_RECEIVES": "${dir:grants}/npm.json",
|
||||
"MESH_GITEA_READY_AUTHORS_FILE": "${dir:mesh-state}/ready-authors",
|
||||
"MESH_GITEA_HOOK_URL": "http://172.17.0.1:3047",
|
||||
"MESH_GITEA_HOOK_PORT": "3047",
|
||||
"MESH_GITEA_HOOK_HOST": "172.17.0.1"
|
||||
"MESH_GITEA_READY_AUTHORS_FILE": "${dir:mesh-state}/ready-authors"
|
||||
}
|
||||
},
|
||||
{
|
||||
|
||||
@@ -15,7 +15,7 @@ import { tmpdir } from "node:os";
|
||||
import { join } from "node:path";
|
||||
|
||||
import { collectTools } from "@novox/mesh-sdk/tools";
|
||||
import { AdminRefused, MINT_RETRY_AFTER_MS, MintedToken, TOKEN_SCOPES } from "../dist/token.js";
|
||||
import { AdminRefused, MintedToken, TOKEN_SCOPES } from "../dist/token.js";
|
||||
import { GiteaClient } from "../dist/client.js";
|
||||
import { getGiteaTools } from "../dist/tools/index.js";
|
||||
|
||||
@@ -29,9 +29,6 @@ interface Forge {
|
||||
mints: number;
|
||||
lastScopes: string[] | null;
|
||||
tokens: Map<string, string>;
|
||||
/** Tokens beyond the one per name `tokens` holds, by id: what two mints racing leave (novox/hq issue 739). */
|
||||
twins: Map<number, { name: string; sha1: string }>;
|
||||
tokenCalls: number;
|
||||
scopesOf: Map<string, string[]>;
|
||||
admins: Map<string, string>;
|
||||
pullState: string;
|
||||
@@ -45,9 +42,6 @@ function fakeForge(): Promise<Forge> {
|
||||
mints: 0,
|
||||
lastScopes: null as string[] | null,
|
||||
tokens: new Map<string, string>(), // name -> value
|
||||
twins: new Map<number, { name: string; sha1: string }>(),
|
||||
idOf: new Map<string, number>(), // value -> id
|
||||
tokenCalls: 0,
|
||||
scopesOf: new Map<string, string[]>(), // value -> scopes, so a route can enforce them like gitea does
|
||||
admins: new Map([[ADMIN, PASSWORD]]),
|
||||
pullState: "open",
|
||||
@@ -79,41 +73,21 @@ function fakeForge(): Promise<Forge> {
|
||||
const tokens = url.pathname.match(/^\/api\/v1\/users\/([^/]+)\/tokens(?:\/([^/]+))?$/);
|
||||
if (tokens) {
|
||||
const user = basic(req);
|
||||
forge.tokenCalls++;
|
||||
if (user === null || user !== decodeURIComponent(tokens[1])) return json(res, 401, { message: "auth required" });
|
||||
const named = (name: string): number =>
|
||||
(forge.tokens.has(name) ? 1 : 0) + [...forge.twins.values()].filter((t) => t.name === name).length;
|
||||
if (req.method === "POST") {
|
||||
const { name, scopes } = await body(req);
|
||||
if (named(name) > 0) return json(res, 400, { message: "token name has already been used" });
|
||||
if (forge.tokens.has(name)) return json(res, 400, { message: "token name has already been used" });
|
||||
forge.mints++;
|
||||
forge.lastScopes = scopes;
|
||||
const sha1 = `minted-${forge.mints}-${Math.random().toString(36).slice(2)}`;
|
||||
forge.tokens.set(name, sha1);
|
||||
forge.idOf.set(sha1, forge.mints);
|
||||
forge.scopesOf.set(sha1, scopes);
|
||||
return json(res, 201, { id: forge.mints, name, sha1, scopes, token_last_eight: sha1.slice(-8) });
|
||||
}
|
||||
if (req.method === "GET" && !tokens[2]) {
|
||||
const page = Number(url.searchParams.get("page") ?? "1");
|
||||
const all = [...[...forge.tokens].map(([name, v]) => ({ id: forge.idOf.get(v) ?? 0, name })),
|
||||
...[...forge.twins].map(([id, t]) => ({ id, name: t.name }))];
|
||||
return json(res, 200, page === 1 ? all : []);
|
||||
}
|
||||
if (req.method === "DELETE" && tokens[2]) {
|
||||
const key = decodeURIComponent(tokens[2]);
|
||||
// gitea: "identified by ID and if not available by name"
|
||||
if (/^\d+$/.test(key)) {
|
||||
const id = Number(key);
|
||||
if (forge.twins.delete(id)) return json(res, 204, null);
|
||||
const hit = [...forge.tokens].find(([, v]) => forge.idOf.get(v) === id);
|
||||
if (!hit) return json(res, 404, { message: "token not found" });
|
||||
forge.tokens.delete(hit[0]);
|
||||
return json(res, 204, null);
|
||||
}
|
||||
if (named(key) > 1) return json(res, 422, { message: `multiple matches for token name '${key}'` });
|
||||
if (!forge.tokens.has(key)) return json(res, 404, { message: "token not found" });
|
||||
forge.tokens.delete(key);
|
||||
const name = decodeURIComponent(tokens[2]);
|
||||
if (!forge.tokens.has(name)) return json(res, 404, { message: "token not found" });
|
||||
forge.tokens.delete(name);
|
||||
return json(res, 204, null);
|
||||
}
|
||||
return json(res, 405, { message: "method not allowed" });
|
||||
@@ -206,9 +180,7 @@ function fakeForge(): Promise<Forge> {
|
||||
get pullState() { return forge.pullState; },
|
||||
get pullTitle() { return forge.pullTitle; },
|
||||
get branchDeleted() { return forge.branchDeleted; },
|
||||
get tokenCalls() { return forge.tokenCalls; },
|
||||
tokens: forge.tokens,
|
||||
twins: forge.twins,
|
||||
scopesOf: forge.scopesOf,
|
||||
admins: forge.admins,
|
||||
close: () => new Promise((r) => server.close(() => r())),
|
||||
@@ -237,9 +209,8 @@ async function delivered(forge: Forge): Promise<{ env: NodeJS.ProcessEnv; file:
|
||||
}
|
||||
|
||||
/** A client as a fresh process would build it: a new source over the kept file, its log captured. */
|
||||
function minted(env: NodeJS.ProcessEnv, logs: string[], now?: () => number): GiteaClient {
|
||||
function minted(env: NodeJS.ProcessEnv, logs: string[]): GiteaClient {
|
||||
const source = new MintedToken({
|
||||
now,
|
||||
url: env.MESH_GITEA_URL!,
|
||||
admin: env.MESH_GITEA_ADMIN_USER!,
|
||||
passwordFile: env.MESH_GITEA_ADMIN_PASSWORD_FILE!,
|
||||
@@ -354,44 +325,6 @@ test("the kept file is gone but the forge still holds a token by that name: repl
|
||||
assert.ok(logs.some((l) => l.includes('already holds a token named "mesh-tools"')), logs.join("\n"));
|
||||
});
|
||||
|
||||
test("two tokens carry the name (two mints raced): each is dropped by its id and one is minted, not a 422 for good (issue 739)", async () => {
|
||||
const { env, logs } = await delivered(forge);
|
||||
await minted(env, logs).listRepos();
|
||||
// What two processes minting at once leave: a second token by the same name, whose value nobody kept.
|
||||
// Two tokens by the name, neither of them the one kept.
|
||||
forge.tokens.clear();
|
||||
forge.twins.set(9001, { name: "mesh-tools", sha1: "raced-first" });
|
||||
forge.twins.set(9002, { name: "mesh-tools", sha1: "raced-second" });
|
||||
|
||||
const repos = await minted(env, logs).listRepos();
|
||||
|
||||
assert.equal(repos.length, 1);
|
||||
assert.equal(forge.twins.size, 0);
|
||||
assert.equal([...forge.tokens.keys()].filter((n) => n === "mesh-tools").length, 1);
|
||||
assert.ok(logs.some((l) => l.includes('holds 2 tokens named "mesh-tools"')), logs.join("\n"));
|
||||
});
|
||||
|
||||
test("a mint that failed is not asked of the forge again within the retry window: the same error, no basic-auth call (issue 739)", async () => {
|
||||
const { env, logs } = await delivered(forge);
|
||||
forge.admins.delete(ADMIN);
|
||||
try {
|
||||
let clock = 5_000_000;
|
||||
const client = minted(env, logs, () => clock);
|
||||
await assert.rejects(client.listRepos(), AdminRefused);
|
||||
const calls = forge.tokenCalls;
|
||||
|
||||
await assert.rejects(client.listRepos(), AdminRefused);
|
||||
await assert.rejects(client.listRepos(), AdminRefused);
|
||||
assert.equal(forge.tokenCalls, calls);
|
||||
|
||||
clock += MINT_RETRY_AFTER_MS;
|
||||
await assert.rejects(client.listRepos(), AdminRefused);
|
||||
assert.equal(forge.tokenCalls, calls + 1);
|
||||
} finally {
|
||||
forge.admins.set(ADMIN, PASSWORD);
|
||||
}
|
||||
});
|
||||
|
||||
test("concurrent first calls share one mint", async () => {
|
||||
const { env, logs } = await delivered(forge);
|
||||
const client = minted(env, logs);
|
||||
@@ -406,8 +339,7 @@ test("the admin account is refused: said plainly, nothing kept, and the next cal
|
||||
const { env, file, logs } = await delivered(forge);
|
||||
forge.admins.delete(ADMIN); // the forge's data came from a predecessor; mesh-admin was never created there
|
||||
try {
|
||||
let clock = 1_000_000;
|
||||
const client = minted(env, logs, () => clock);
|
||||
const client = minted(env, logs);
|
||||
const before = forge.mints;
|
||||
|
||||
await assert.rejects(client.listRepos(), (err: unknown) => {
|
||||
@@ -422,9 +354,8 @@ test("the admin account is refused: said plainly, nothing kept, and the next cal
|
||||
assert.equal(forge.mints, before);
|
||||
await assert.rejects(stat(file), /ENOENT/);
|
||||
|
||||
// The account appears (the operator created it): the next call after the retry window mints and works.
|
||||
// The account appears (the operator created it): the very next call mints and works.
|
||||
forge.admins.set(ADMIN, PASSWORD);
|
||||
clock += MINT_RETRY_AFTER_MS;
|
||||
assert.equal((await client.listRepos()).length, 1);
|
||||
assert.equal(forge.mints, before + 1);
|
||||
} finally {
|
||||
|
||||
@@ -1,120 +0,0 @@
|
||||
import assert from "node:assert/strict";
|
||||
import { test } from "node:test";
|
||||
import { createHmac } from "node:crypto";
|
||||
import type { AddressInfo } from "node:net";
|
||||
|
||||
// **The forge says every event on the bus** (novox/hq issue 729): a signed delivery is published once and acted on,
|
||||
// an unsigned or unverified one is refused, and an event and the poll seeing the same head announce it once.
|
||||
|
||||
const secret = "s".repeat(64);
|
||||
const sign = (body: string, key = secret) => createHmac("sha256", key).update(body).digest("hex");
|
||||
const opened = JSON.stringify({ action: "opened", number: 7, pull_request: { number: 7, title: "t", user: { login: "jo" },
|
||||
head: { ref: "feat/x", sha: "a".repeat(40) }, base: { ref: "main" }, merged: false },
|
||||
repository: { name: "hq", full_name: "novox/hq", owner: { login: "novox" } }, sender: { login: "jo" } });
|
||||
|
||||
async function serve(published: { name: string }[], acted: { name: string }[]) {
|
||||
const { receiver, HOOK_PATH } = await import("../webhook.ts");
|
||||
const server = receiver({ secret, port: 0, log: () => {},
|
||||
publish: async (e) => { published.push(e); }, act: async (e) => { acted.push(e); } });
|
||||
await new Promise<void>((r) => server.listen(0, "127.0.0.1", () => r()));
|
||||
const url = `http://127.0.0.1:${(server.address() as AddressInfo).port}${HOOK_PATH}`;
|
||||
return { server, url };
|
||||
}
|
||||
|
||||
test("a signed event is published on the bus and acted on once", async () => {
|
||||
const published: { name: string }[] = [], acted: { name: string }[] = [];
|
||||
const { server, url } = await serve(published, acted);
|
||||
try {
|
||||
const headers = { "X-Gitea-Event": "pull_request", "X-Gitea-Signature": sign(opened), "X-Gitea-Delivery": "d-1" };
|
||||
const res = await fetch(url, { method: "POST", body: opened, headers });
|
||||
assert.equal(res.status, 200);
|
||||
// The same delivery sent again is answered and published nowhere.
|
||||
const again = await fetch(url, { method: "POST", body: opened, headers });
|
||||
assert.equal(again.status, 200);
|
||||
await server.idle();
|
||||
assert.deepEqual(published.map((e) => e.name), ["forge.pull-request.opened"]);
|
||||
assert.deepEqual(acted.map((e) => e.name), ["forge.pull-request.opened"]);
|
||||
} finally { server.close(); }
|
||||
});
|
||||
|
||||
test("an unsigned event, or one whose signature does not verify, is refused and published nowhere", async () => {
|
||||
const published: { name: string }[] = [], acted: { name: string }[] = [];
|
||||
const { server, url } = await serve(published, acted);
|
||||
try {
|
||||
const unsigned = await fetch(url, { method: "POST", body: opened, headers: { "X-Gitea-Event": "pull_request" } });
|
||||
assert.equal(unsigned.status, 401);
|
||||
assert.match(await unsigned.text(), /unsigned/);
|
||||
const forged = await fetch(url, { method: "POST", body: opened,
|
||||
headers: { "X-Gitea-Event": "pull_request", "X-Gitea-Signature": sign(opened, "another secret") } });
|
||||
assert.equal(forged.status, 401);
|
||||
assert.match(await forged.text(), /does not verify/);
|
||||
const big = "x".repeat(5 * 1024 * 1024 + 1);
|
||||
const large = await fetch(url, { method: "POST", body: big,
|
||||
headers: { "X-Gitea-Event": "push", "X-Gitea-Signature": sign(big) } });
|
||||
assert.equal(large.status, 413);
|
||||
await server.idle();
|
||||
assert.deepEqual(published, []);
|
||||
assert.deepEqual(acted, []);
|
||||
} finally { server.close(); }
|
||||
});
|
||||
|
||||
test("an event and the poll seeing the same head announce it once", async () => {
|
||||
const { claimHead } = await import("../webhook.ts");
|
||||
const heads: Record<string, string> = {};
|
||||
const sha = "a".repeat(40);
|
||||
assert.equal(claimHead(heads, "novox/hq#7", sha), true, "the event claims the head and announces it");
|
||||
assert.equal(claimHead(heads, "novox/hq#7", sha), false, "the poll finds it taken");
|
||||
assert.equal(claimHead(heads, "novox/hq#7", "b".repeat(40)), true, "a new head is announced again");
|
||||
});
|
||||
|
||||
test("each forge event is named in the module's stream and trimmed", async () => {
|
||||
const { eventOf } = await import("../webhook.ts");
|
||||
const merged = eventOf("pull_request", { action: "closed", pull_request: { number: 3, merged: true, merge_commit_sha: "c".repeat(40),
|
||||
head: { sha: "a".repeat(40) }, base: { ref: "main" } }, repository: { name: "hq", owner: { login: "novox" } } });
|
||||
assert.equal(merged.name, "forge.pull-request.merged");
|
||||
assert.equal(merged.body.merge_commit_sha, "c".repeat(40));
|
||||
assert.equal(eventOf("pull_request", { action: "label_updated", pull_request: { number: 3 }, label: { name: "mesh/ready" } }).name,
|
||||
"forge.pull-request.labeled");
|
||||
const comment = eventOf("issue_comment", { action: "created", is_pull: true, issue: { number: 3 },
|
||||
comment: { id: 9, body: "ready at abc", user: { login: "jo" } } });
|
||||
assert.equal(comment.name, "forge.pull-request.commented");
|
||||
assert.equal(comment.body.comment, "ready at abc");
|
||||
assert.equal(eventOf("push", { ref: "refs/heads/main", after: "d".repeat(40) }).name, "forge.push");
|
||||
assert.equal(eventOf("pull_request_review_approved", {}).name, "forge.other");
|
||||
});
|
||||
|
||||
test("the system webhook is made when none points at the receiver, and corrected when one does", async () => {
|
||||
const { ensureSystemHook } = await import("../webhook.ts");
|
||||
const calls: string[] = [];
|
||||
let hooks: unknown[] = [];
|
||||
const admin = { call: async (method: string, path: string, body?: any) => {
|
||||
calls.push(`${method} ${path}`);
|
||||
if (method === "GET") return { status: 200, body: hooks };
|
||||
if (method === "POST") { assert.ok(body.events.includes("pull_request") && body.events.includes("push")); assert.equal(body.config.secret, secret); return { status: 201, body: {} }; }
|
||||
return { status: 200, body: {} };
|
||||
} };
|
||||
assert.equal(await ensureSystemHook(admin, "http://h:3047/forge-event", secret), "created");
|
||||
hooks = [{ id: 4, config: { url: "http://h:3047/forge-event" } }];
|
||||
assert.equal(await ensureSystemHook(admin, "http://h:3047/forge-event", secret), "corrected");
|
||||
assert.ok(calls.includes("PATCH /admin/hooks/4"));
|
||||
});
|
||||
|
||||
test("the polls look every 30 s until events are heard, every five minutes while they are, and 30 s again on a miss", async () => {
|
||||
const { Pace } = await import("../webhook.ts");
|
||||
let now = 1_000_000;
|
||||
const said: string[] = [];
|
||||
const pace = new Pace(() => now, (l) => said.push(l));
|
||||
assert.equal(pace.every(), 30_000, "no delivery yet");
|
||||
pace.heard();
|
||||
assert.equal(pace.every(), 300_000, "a delivery heard");
|
||||
now += 11 * 60_000;
|
||||
assert.equal(pace.every(), 30_000, "ten minutes without a delivery");
|
||||
pace.heard();
|
||||
now += 1000;
|
||||
pace.miss("novox/hq#7 at aaaaaaaa");
|
||||
assert.equal(pace.every(), 30_000, "the poll found what no event said");
|
||||
assert.match(said.join("\n"), /MISSED novox\/hq#7/);
|
||||
now += 1000;
|
||||
pace.heard();
|
||||
assert.equal(pace.every(), 300_000, "the next delivery");
|
||||
});
|
||||
+4
-60
@@ -97,25 +97,13 @@ export interface MintedTokenOptions {
|
||||
readonly scopes?: readonly string[];
|
||||
readonly log?: (line: string) => void;
|
||||
readonly fetch?: typeof fetch;
|
||||
/** The clock, in milliseconds; a test sets it. */
|
||||
readonly now?: () => number;
|
||||
}
|
||||
|
||||
/**
|
||||
* **How long a failed mint is not tried again** (novox/hq issue 739). Every call that finds no token mints, and a
|
||||
* mint is two or three calls over basic auth, each one hashing the admin password on the forge. A mint that keeps
|
||||
* failing, once per tool call, is a loop the forge pays for in CPU: on 2026-10-14 it ran several times a minute
|
||||
* on the control-node while it was already overloaded. Within this window the last failure is answered again,
|
||||
* without asking the forge.
|
||||
*/
|
||||
export const MINT_RETRY_AFTER_MS = 30_000;
|
||||
|
||||
/** The token the module mints for itself, kept in its state and renewed when the forge rejects it. */
|
||||
export class MintedToken implements TokenSource {
|
||||
private held: string | null = null;
|
||||
private readFile = false;
|
||||
private inflight: Promise<string> | null = null;
|
||||
private failed: { at: number; error: unknown } | null = null;
|
||||
private readonly name: string;
|
||||
private readonly scopes: readonly string[];
|
||||
private readonly log: (line: string) => void;
|
||||
@@ -126,11 +114,8 @@ export class MintedToken implements TokenSource {
|
||||
this.scopes = opts.scopes ?? TOKEN_SCOPES;
|
||||
this.log = opts.log ?? ((line) => console.log(`[gitea] ${line}`));
|
||||
this.fetchImpl = opts.fetch ?? fetch;
|
||||
this.now = opts.now ?? Date.now;
|
||||
}
|
||||
|
||||
private readonly now: () => number;
|
||||
|
||||
/**
|
||||
* Build from the runtime's environment: the forge's URL, the admin login and password file the
|
||||
* manifest hands the runtime, and the module's state directory (MESH_GITEA_STATE_DIR, a directory
|
||||
@@ -141,10 +126,7 @@ export class MintedToken implements TokenSource {
|
||||
const opts = MintedToken.optionsFromEnv(url, env);
|
||||
// One source per kept file in a process. The watcher and the tools entrypoint both build a
|
||||
// client in the same runtime; two sources over one file would each renew on a 401 and drop the
|
||||
// other's token by name, forever. Shared, a renewal is one renewal. Across processes it is not shared: the
|
||||
// runtime serves the watcher (index.js) and the tools (tools/index.js) as separate bundles, so two sources over
|
||||
// one file do exist. renew() reads the kept file before it mints, and a race that still leaves two tokens by
|
||||
// the name is undone by dropping each by its id (doMint; novox/hq issue 739).
|
||||
// other's token by name, forever. Shared, a renewal is one renewal.
|
||||
const shared = MintedToken.shared.get(opts.file);
|
||||
if (shared) return shared;
|
||||
const source = new MintedToken(opts);
|
||||
@@ -181,7 +163,7 @@ export class MintedToken implements TokenSource {
|
||||
}
|
||||
if (kept !== null) return this.mint("the kept token was minted with other scopes than the tools take now — minting one with them");
|
||||
}
|
||||
return this.mint(this.failed ? "no token held since the last mint failed — minting one" : "no token kept — minting one");
|
||||
return this.mint("no token kept — minting one");
|
||||
}
|
||||
|
||||
async renew(rejected: string): Promise<string> {
|
||||
@@ -206,19 +188,8 @@ export class MintedToken implements TokenSource {
|
||||
/** One mint at a time: concurrent first calls share it, rather than each minting its own. */
|
||||
private mint(why: string): Promise<string> {
|
||||
if (this.inflight === null) {
|
||||
// A mint that failed a moment ago is not asked of the forge again yet (novox/hq issue 739): its error stands.
|
||||
if (this.failed !== null && this.now() - this.failed.at < MINT_RETRY_AFTER_MS) return Promise.reject(this.failed.error);
|
||||
this.log(why);
|
||||
this.inflight = this.doMint().then(
|
||||
(token) => {
|
||||
this.failed = null;
|
||||
return token;
|
||||
},
|
||||
(err) => {
|
||||
this.failed = { at: this.now(), error: err };
|
||||
throw err;
|
||||
},
|
||||
).finally(() => {
|
||||
this.inflight = this.doMint().finally(() => {
|
||||
this.inflight = null;
|
||||
});
|
||||
}
|
||||
@@ -290,12 +261,7 @@ export class MintedToken implements TokenSource {
|
||||
// went while the forge's data stayed. It is ours to replace: drop it by name and mint again.
|
||||
this.log(`the forge already holds a token named "${this.name}" — replacing it`);
|
||||
const dropped = await call("DELETE", `/${encodeURIComponent(this.name)}`);
|
||||
if (dropped.status === 422) {
|
||||
// More than one token carries the name (novox/hq issue 739): two processes minted at once, and the forge's
|
||||
// check that a name is unused is not atomic. The forge refuses a delete by such a name for good, so every
|
||||
// one of them is dropped by its id — all are ours, and none of their values is kept.
|
||||
await this.dropEveryNamed(call, dropped);
|
||||
} else if (dropped.status !== 204 && dropped.status !== 404) {
|
||||
if (dropped.status !== 204 && dropped.status !== 404) {
|
||||
throw new Error(`Gitea DELETE /users/${this.opts.admin}/tokens/${this.name}: ${dropped.status} ${detail(dropped.body)}`);
|
||||
}
|
||||
res = await call("POST", "", { name: this.name, scopes: this.scopes });
|
||||
@@ -311,28 +277,6 @@ export class MintedToken implements TokenSource {
|
||||
this.log(`minted a token for "${this.opts.admin}" (${this.scopes.join(", ")}), kept at ${this.opts.file}`);
|
||||
return token;
|
||||
}
|
||||
|
||||
/** Drop, by id, every token of the admin account that carries this name; the forge listed them, page by page. */
|
||||
private async dropEveryNamed(call: (method: string, path?: string) => Promise<{ status: number; body: any }>,
|
||||
refused: { status: number; body: any }): Promise<void> {
|
||||
const ids: number[] = [];
|
||||
for (let page = 1; page <= 20; page++) {
|
||||
const listed = await call("GET", `?page=${page}&limit=50`);
|
||||
if (listed.status !== 200 || !Array.isArray(listed.body)) {
|
||||
throw new Error(`Gitea DELETE /users/${this.opts.admin}/tokens/${this.name}: ${refused.status} ${detail(refused.body)}; ` +
|
||||
`and its tokens could not be listed to drop them one by one: ${listed.status} ${detail(listed.body)}`);
|
||||
}
|
||||
for (const t of listed.body) if (t?.name === this.name && Number.isInteger(t?.id)) ids.push(t.id);
|
||||
if (listed.body.length < 50) break;
|
||||
}
|
||||
this.log(`the forge holds ${ids.length} tokens named "${this.name}" — dropping each by its id`);
|
||||
for (const id of ids) {
|
||||
const dropped = await call("DELETE", `/${id}`);
|
||||
if (dropped.status !== 204 && dropped.status !== 404) {
|
||||
throw new Error(`Gitea DELETE /users/${this.opts.admin}/tokens/${id}: ${dropped.status} ${detail(dropped.body)}`);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
function detail(body: unknown): string {
|
||||
|
||||
@@ -1,314 +0,0 @@
|
||||
// **The forge says every event on the bus** (novox/hq issue 729; to-be 10: an event makes it fast, nothing makes it
|
||||
// necessary). The module keeps one system webhook on its forge for every event, pointing at its own receiver, signed
|
||||
// with a secret it mints and keeps beside its token; the receiver refuses whatever is unsigned or does not verify, and
|
||||
// publishes each event, trimmed, on the module's events as forge.<kind>.<action> (forge.pull-request.opened, forge.push, …),
|
||||
// on the shared EVENTS path its other events take; its own stream (ADR 0363) is a follow-up once issue 725 ships.
|
||||
// The polls stay as the safety net, slowed; the announced-heads and announced-merges records keep anything from
|
||||
// being announced twice, whichever path saw it first.
|
||||
//
|
||||
// The secret is never logged: lines say it was minted or reused and where it is kept, never what it is.
|
||||
|
||||
import { createHmac, randomBytes, timingSafeEqual } from "node:crypto";
|
||||
import { chmodSync, mkdirSync, readFileSync, renameSync, writeFileSync } from "node:fs";
|
||||
import { createServer, type Server } from "node:http";
|
||||
import { dirname } from "node:path";
|
||||
|
||||
/** The name the receiver answers on, and the port it listens on unless the manifest says otherwise. */
|
||||
export const HOOK_PATH = "/forge-event";
|
||||
export const HOOK_PORT = 3047;
|
||||
|
||||
/** The secret the forge signs with: read from its 0600 file, or minted and kept there. */
|
||||
export function keptSecret(file: string, log: (line: string) => void = (l) => console.log(`[gitea] ${l}`)): string {
|
||||
try {
|
||||
const kept = readFileSync(file, "utf8").trim();
|
||||
if (kept.length >= 32) {
|
||||
log(`reusing the webhook secret kept at ${file}`);
|
||||
return kept;
|
||||
}
|
||||
} catch (err) {
|
||||
if ((err as NodeJS.ErrnoException).code !== "ENOENT") throw err;
|
||||
}
|
||||
const secret = randomBytes(32).toString("hex");
|
||||
mkdirSync(dirname(file), { recursive: true, mode: 0o700 });
|
||||
const tmp = `${file}.tmp`;
|
||||
writeFileSync(tmp, secret + "\n", { mode: 0o600 });
|
||||
chmodSync(tmp, 0o600);
|
||||
renameSync(tmp, file);
|
||||
log(`minted a webhook secret, kept at ${file}`);
|
||||
return secret;
|
||||
}
|
||||
|
||||
/** Whether a body carries the forge's HMAC-SHA256 signature (X-Gitea-Signature, hex) under the secret. */
|
||||
export function verified(secret: string, body: Buffer | string, signature: string | undefined): boolean {
|
||||
if (!secret || !signature || !/^[0-9a-f]{64}$/i.test(signature.trim())) return false;
|
||||
const want = createHmac("sha256", secret).update(body).digest();
|
||||
const got = Buffer.from(signature.trim(), "hex");
|
||||
return got.length === want.length && timingSafeEqual(got, want);
|
||||
}
|
||||
|
||||
/** One forge event as the bus carries it: its name inside the stream, and what is needed of it — no secrets. */
|
||||
export interface ForgeEvent {
|
||||
name: string;
|
||||
body: Record<string, unknown>;
|
||||
}
|
||||
|
||||
/** Every event name the module emits for the forge, without its forge. prefix (module.json `emits`). */
|
||||
export const EVENTS: readonly string[] = ["push", "create", "delete", "fork", "status", "pull-request.opened", "pull-request.reopened", "pull-request.synchronized", "pull-request.edited", "pull-request.closed", "pull-request.merged", "pull-request.labeled", "pull-request.unlabeled", "pull-request.assigned", "pull-request.unassigned", "pull-request.reviewed", "pull-request.review-requested", "pull-request.commented", "pull-request.changed", "issue-comment.created", "issue-comment.edited", "issue-comment.deleted", "issues.opened", "issues.reopened", "issues.closed", "issues.edited", "issues.labeled", "issues.unlabeled", "issues.assigned", "issues.unassigned", "issues.changed", "release.published", "release.updated", "release.deleted", "release.changed", "repository.created", "repository.deleted", "repository.changed", "wiki.changed", "other"];
|
||||
|
||||
/** The families the module's stream carries (module.json `streams`); any other kind is said as other.<kind>; underscores become hyphens. */
|
||||
export const FAMILIES = ["push", "create", "delete", "fork", "status", "pull_request", "issue_comment", "issues",
|
||||
"release", "repository", "wiki"] as const;
|
||||
|
||||
const ACTIONS: Record<string, string> = { label_updated: "labeled", label_cleared: "unlabeled", synchronized: "synchronized" };
|
||||
|
||||
function str(v: unknown): string | undefined {
|
||||
return typeof v === "string" && v ? v : undefined;
|
||||
}
|
||||
function num(v: unknown): number | undefined {
|
||||
return typeof v === "number" ? v : undefined;
|
||||
}
|
||||
function login(u: any): string | undefined {
|
||||
return str(u?.login) ?? str(u?.username);
|
||||
}
|
||||
|
||||
/**
|
||||
* The event a forge delivery is: its kind (X-Gitea-Event) and the payload's action, trimmed to ids, repository, refs,
|
||||
* shas, author, labels and a comment's body (a ready mark is one, ADR 0364). A merge is pull-request.merged, a comment on
|
||||
* a pull request pull-request.commented.
|
||||
*/
|
||||
export function eventOf(kind: string, payload: any): ForgeEvent {
|
||||
// The bus's event names take lower-case letters, digits and hyphens: the forge's underscores become hyphens.
|
||||
const event = forgeEventOf(kind, payload);
|
||||
const name = event.name.toLowerCase().replace(/_/g, "-").replace(/[^a-z0-9.-]/g, "-");
|
||||
// Published on the module's events as it publishes its others, each name declared in its `emits`: one the manifest
|
||||
// does not name is forge.other, saying what the forge called it. Its own stream (ADR 0363) follows issue 725.
|
||||
if (EVENTS.includes(name)) return { name: `forge.${name}`, body: event.body };
|
||||
return { name: "forge.other", body: { ...event.body, kind, said: name } };
|
||||
}
|
||||
|
||||
function forgeEventOf(said: string, payload: any): ForgeEvent {
|
||||
// The forge names some kinds finer than their family (pull_request_label, pull_request_sync): read by the payload.
|
||||
let kind = said;
|
||||
if (said.startsWith("pull_request") && payload?.pull_request) kind = "pull_request";
|
||||
if ((said.startsWith("issue") || said === "pull_request_comment") && payload?.comment) kind = "issue_comment";
|
||||
if (said.startsWith("issue") && !payload?.comment && payload?.issue) kind = "issues";
|
||||
const repo = payload?.repository ?? {};
|
||||
const base: Record<string, unknown> = {
|
||||
owner: login(repo.owner), repo: str(repo.name), full_name: str(repo.full_name), clone_url: str(repo.clone_url),
|
||||
sender: login(payload?.sender),
|
||||
};
|
||||
const family = (FAMILIES as readonly string[]).includes(kind) ? kind : `other`;
|
||||
let action = str(payload?.action);
|
||||
const pull = payload?.pull_request;
|
||||
if (kind === "push") {
|
||||
return { name: "push", body: { ...base, ref: str(payload.ref), before: str(payload.before), after: str(payload.after),
|
||||
commits: Array.isArray(payload.commits) ? payload.commits.length : undefined, pusher: login(payload.pusher) } };
|
||||
}
|
||||
if (kind === "pull_request" && pull) {
|
||||
if (action === "closed" && pull.merged) action = "merged";
|
||||
action = ACTIONS[action ?? ""] ?? action ?? "changed";
|
||||
return { name: `pull_request.${action}`, body: { ...base, number: num(pull.number ?? payload.number),
|
||||
title: str(pull.title), author: login(pull.user), head: str(pull.head?.ref), head_sha: str(pull.head?.sha),
|
||||
base: str(pull.base?.ref), merged: !!pull.merged, merge_commit_sha: str(pull.merge_commit_sha),
|
||||
merged_at: str(pull.merged_at), html_url: str(pull.html_url),
|
||||
labels: Array.isArray(pull.labels) ? pull.labels.map((l: any) => str(l?.name)).filter(Boolean) : undefined,
|
||||
label: str(payload.label?.name) } };
|
||||
}
|
||||
if (kind === "issue_comment") {
|
||||
const issue = payload.issue ?? {};
|
||||
const onPull = !!payload.is_pull || !!issue.pull_request;
|
||||
return { name: onPull ? "pull_request.commented" : `issue_comment.${action ?? "created"}`, body: { ...base,
|
||||
number: num(issue.number), comment_id: num(payload.comment?.id), author: login(payload.comment?.user),
|
||||
comment: str(payload.comment?.body), action } };
|
||||
}
|
||||
if (kind === "create" || kind === "delete") {
|
||||
return { name: kind, body: { ...base, ref: str(payload.ref), ref_type: str(payload.ref_type), sha: str(payload.sha) } };
|
||||
}
|
||||
if (kind === "status") {
|
||||
return { name: "status", body: { ...base, sha: str(payload.sha), context: str(payload.context), state: str(payload.state) } };
|
||||
}
|
||||
const name = family === "other" ? `other.${kind.replace(/[^a-z0-9_-]/gi, "_") || "unknown"}` :
|
||||
["fork"].includes(family) ? family : `${family}.${action ?? "changed"}`;
|
||||
return { name, body: { ...base, number: num(payload?.number ?? payload?.issue?.number), action } };
|
||||
}
|
||||
|
||||
export interface ReceiverOptions {
|
||||
secret: string;
|
||||
port: number;
|
||||
host?: string;
|
||||
/** Publishes one event on the bus. */
|
||||
publish: (event: ForgeEvent) => Promise<void>;
|
||||
/** Acts on it at once, beside publishing it (the announcer). */
|
||||
act?: (event: ForgeEvent) => Promise<void>;
|
||||
/** Told of every verified delivery, so the polls know events are heard. */
|
||||
heard?: () => void;
|
||||
log?: (line: string) => void;
|
||||
}
|
||||
|
||||
/** The most a delivery may carry: the forge's push of many commits stays well under it. */
|
||||
export const MAX_BODY = 5 * 1024 * 1024;
|
||||
/** How many deliveries' ids are remembered, so a delivery sent again is published once. */
|
||||
export const REMEMBERED = 2000;
|
||||
|
||||
/**
|
||||
* The receiver. A request without a signature is refused before its body is read, a body over MAX_BODY is refused, and
|
||||
* one whose signature does not verify is refused, each with 401 or 413 and a line saying so. A verified delivery is
|
||||
* answered 200 at once and queued: it is published and acted on after the reply, one at a time, so the forge's
|
||||
* delivery timeout is never reached. A delivery whose X-Gitea-Delivery id was already heard is answered and dropped.
|
||||
*/
|
||||
export function receiver(opts: ReceiverOptions): Server & { idle(): Promise<void> } {
|
||||
const log = opts.log ?? ((l) => console.log(`[gitea] ${l}`));
|
||||
const seen = new Set<string>();
|
||||
let queue: Promise<void> = Promise.resolve();
|
||||
const server = createServer((req, res) => {
|
||||
if (req.method !== "POST" || (req.url ?? "").split("?")[0] !== HOOK_PATH) {
|
||||
res.statusCode = 404;
|
||||
return res.end();
|
||||
}
|
||||
const from = req.socket.remoteAddress ?? "?";
|
||||
const signature = String(req.headers["x-gitea-signature"] ?? "").trim();
|
||||
if (!signature) {
|
||||
log(`refused a forge event from ${from}: it is unsigned`);
|
||||
res.writeHead(401, { Connection: "close" });
|
||||
res.end("unsigned\n");
|
||||
req.destroy();
|
||||
return;
|
||||
}
|
||||
if (Number(req.headers["content-length"] ?? 0) > MAX_BODY) {
|
||||
log(`refused a forge event from ${from}: its body is over ${MAX_BODY} bytes`);
|
||||
res.writeHead(413, { Connection: "close" });
|
||||
res.end("too large\n");
|
||||
req.destroy();
|
||||
return;
|
||||
}
|
||||
const chunks: Buffer[] = [];
|
||||
let size = 0;
|
||||
let over = false;
|
||||
req.on("data", (c: Buffer) => {
|
||||
size += c.length;
|
||||
if (size > MAX_BODY) {
|
||||
if (!over) {
|
||||
over = true;
|
||||
log(`refused a forge event from ${from}: its body is over ${MAX_BODY} bytes`);
|
||||
res.writeHead(413, { Connection: "close" });
|
||||
res.end("too large\n");
|
||||
req.destroy();
|
||||
}
|
||||
} else chunks.push(c);
|
||||
});
|
||||
req.on("end", () => {
|
||||
if (over) return;
|
||||
const body = Buffer.concat(chunks);
|
||||
if (!verified(opts.secret, body, signature)) {
|
||||
log(`refused a forge event from ${from}: its signature does not verify`);
|
||||
res.statusCode = 401;
|
||||
return res.end("signature does not verify\n");
|
||||
}
|
||||
let payload: unknown;
|
||||
try {
|
||||
payload = JSON.parse(body.toString("utf8"));
|
||||
} catch {
|
||||
res.statusCode = 400;
|
||||
return res.end("not JSON\n");
|
||||
}
|
||||
opts.heard?.();
|
||||
const delivery = String(req.headers["x-gitea-delivery"] ?? "");
|
||||
if (delivery && seen.has(delivery)) {
|
||||
res.statusCode = 200;
|
||||
return res.end("already heard\n");
|
||||
}
|
||||
if (delivery) {
|
||||
seen.add(delivery);
|
||||
if (seen.size > REMEMBERED) seen.delete(seen.values().next().value as string);
|
||||
}
|
||||
const event = eventOf(String(req.headers["x-gitea-event"] ?? "other"), payload);
|
||||
res.statusCode = 200;
|
||||
res.end("queued\n");
|
||||
queue = queue.then(async () => {
|
||||
try {
|
||||
await opts.publish(event);
|
||||
if (opts.act) await opts.act(event);
|
||||
} catch (err) {
|
||||
log(`a forge event ${event.name} could not be published or acted on; the polls catch it — ${err instanceof Error ? err.message : err}`);
|
||||
}
|
||||
});
|
||||
});
|
||||
});
|
||||
return Object.assign(server, { idle: () => queue });
|
||||
}
|
||||
|
||||
/** What ensuring the system webhook needs of the forge's administration (GiteaAdmin's login). */
|
||||
export interface HookAdmin {
|
||||
call(method: string, path: string, body?: unknown): Promise<{ status: number; body: any }>;
|
||||
}
|
||||
|
||||
/**
|
||||
* **How fast the polls look** (issue 729): every five minutes only while the forge's events are heard — a verified
|
||||
* delivery within the last ten minutes, and none missed since. Until the first delivery, after ten minutes without one,
|
||||
* or once the poll found something no event said, every thirty seconds.
|
||||
*/
|
||||
export class Pace {
|
||||
private last = 0;
|
||||
private missed = 0;
|
||||
private readonly now: () => number;
|
||||
private readonly log: (l: string) => void;
|
||||
constructor(now: () => number = Date.now, log: (l: string) => void = (l) => console.error(`[gitea] ${l}`)) {
|
||||
this.now = now;
|
||||
this.log = log;
|
||||
}
|
||||
heard(): void {
|
||||
this.last = this.now();
|
||||
}
|
||||
/** The poll found a head or merge no forge event announced. */
|
||||
miss(what: string): void {
|
||||
if (this.hearing()) this.log(`THE FORGE'S EVENTS MISSED ${what}: the poll found it, so the polls look every 30 s again until the next event`);
|
||||
this.missed = this.now();
|
||||
}
|
||||
hearing(): boolean {
|
||||
const t = this.now();
|
||||
return this.last > 0 && t - this.last < 10 * 60_000 && this.last > this.missed;
|
||||
}
|
||||
every(): number {
|
||||
return this.hearing() ? 5 * 60_000 : 30_000;
|
||||
}
|
||||
}
|
||||
|
||||
/** Every event family the forge (1.27) can send a webhook. */
|
||||
export const HOOK_EVENTS = ["create", "delete", "fork", "push", "issues", "issue_comment", "pull_request",
|
||||
"pull_request_review_request", "wiki", "repository", "release", "package", "status", "workflow_run", "workflow_job"];
|
||||
|
||||
/**
|
||||
* **One system webhook, for every event, at the receiver** (issue 729): created when none points there, corrected
|
||||
* when one does but differs (events, active, content type), and its secret set again each start — the forge never
|
||||
* hands a secret back, so a corrected secret cannot be told from a kept one. Answers what it did.
|
||||
*/
|
||||
export async function ensureSystemHook(admin: HookAdmin, url: string, secret: string): Promise<"created" | "corrected"> {
|
||||
// POST /admin/hooks makes a system webhook, and its options take no flag for it (the forge's 1.27.3 API schema). The
|
||||
// forge has no "every event" word: each family is named, and a family name (issues, pull_request) takes all of it.
|
||||
const want = { type: "gitea", active: true, events: HOOK_EVENTS, branch_filter: "*",
|
||||
config: { url, content_type: "json", secret } };
|
||||
const listed = await admin.call("GET", "/admin/hooks?type=system&limit=50");
|
||||
if (listed.status !== 200) throw new Error(`the forge's system webhooks could not be listed: ${listed.status}`);
|
||||
const hooks: any[] = Array.isArray(listed.body) ? listed.body : [];
|
||||
const ours = hooks.filter((h) => h?.config?.url === url);
|
||||
if (ours.length === 0) {
|
||||
const made = await admin.call("POST", "/admin/hooks", want);
|
||||
if (made.status !== 201 && made.status !== 200) throw new Error(`the forge's system webhook could not be made: ${made.status}`);
|
||||
return "created";
|
||||
}
|
||||
for (const extra of ours.slice(1)) await admin.call("DELETE", `/admin/hooks/${extra.id}`);
|
||||
const fixed = await admin.call("PATCH", `/admin/hooks/${ours[0].id}`, { active: true, events: HOOK_EVENTS, branch_filter: "*",
|
||||
config: { url, content_type: "json", secret } });
|
||||
if (fixed.status !== 200) throw new Error(`the forge's system webhook could not be corrected: ${fixed.status}`);
|
||||
return "corrected";
|
||||
}
|
||||
|
||||
/**
|
||||
* **Claim a head for announcing, once** (issue 729): true when `sha` is not already the head recorded for `key`, which
|
||||
* it then is. The poll and the forge's event both claim before they announce, so whichever comes second finds it taken.
|
||||
*/
|
||||
export function claimHead(heads: Record<string, string>, key: string, sha: string): boolean {
|
||||
if (!sha || heads[key] === sha) return false;
|
||||
heads[key] = sha;
|
||||
return true;
|
||||
}
|
||||
@@ -0,0 +1,58 @@
|
||||
# mesh-mail — the operator's mail accounts, watched and said on the bus
|
||||
|
||||
One module holds the operator's mail accounts, watches them, and says each change of a message on the bus for the
|
||||
modules the operator grants it to (novox/hq ADR 0359, research 049 document 02). One instance, on the control-node.
|
||||
This is the first stage: accounts of the kind `imap`, read-only. Nothing here flags, moves, deletes, drafts or sends.
|
||||
|
||||
## What it does
|
||||
|
||||
- **Watches.** For each account it keeps one connection: IDLE on the inbox, ended and begun again every two minutes
|
||||
(which proves the server still answers), and a look at every other folder every five minutes. Folders are opened
|
||||
read-only and bodies are peeked, so watching never marks a message seen.
|
||||
- **Indexes.** One row per message and folder in its own database (`postgres-database`): metadata, flags, the text
|
||||
for search, and the attachments' references. Each attachment is kept once, by its SHA-256, in its bucket
|
||||
(`s3-bucket`, from `minio`).
|
||||
- **Says each change.** A message added, its flags changed, or removed is said as the event
|
||||
`mail.<account>.message.<added|changed|removed>`: account, folder, uid, message id, date, sender, recipients,
|
||||
subject, size, flags and attachment references; never the body, never an attachment's bytes. The events go to the
|
||||
module's own stream `MAIL`, kept seven days, never to the shared EVENTS stream (novox/hq ADR 0363); a module hears an
|
||||
account's part only when its `consumes` names mesh-mail and the setting `grants` names it for that account. The first look at a
|
||||
folder indexes what is there and says nothing: the bus hears what changed since the watch began. The keeper writes
|
||||
each event in the same step as the change it says; the bundle says it through the runtime and removes it once the
|
||||
bus took it.
|
||||
|
||||
## Its settings, both trusted
|
||||
|
||||
`accounts` — per account name (a lower-case token, which names the account on the bus and its secret):
|
||||
|
||||
```json
|
||||
{"mesh": {"kind": "imap", "address": "<the address>",
|
||||
"incoming": {"host": "<the IMAP server>", "port": 993, "security": "tls"},
|
||||
"outgoing": {"host": "<the SMTP server>", "port": 465, "security": "tls"},
|
||||
"folders": ["optional: the folders watched besides the inbox; every folder when left out"]}}
|
||||
```
|
||||
|
||||
`outgoing` is read and never used in this stage. A second `imap` account on another server is one more entry and one
|
||||
more password; no code changes. `user` signs in under another name than the address.
|
||||
|
||||
`grants` — per account, who may read its mail: a module's name (on every machine), `<machine>.<module>`, or
|
||||
`person.<name>`. Nobody by default. It restricts the stream `MAIL` on the bus (a person's entry grants nothing there)
|
||||
and the tools `mail_search` and `mail_read`.
|
||||
|
||||
The password of an account is the own secret `mail-credential-<account>`, given at the desk; the mesh never makes one.
|
||||
|
||||
## Its tools
|
||||
|
||||
| Tool | Answers |
|
||||
|---|---|
|
||||
| `mail_accounts` | each account's kind, state and why, when its server last answered, its folders; the address only to a caller granted it |
|
||||
| `mail_search` | messages of the accounts the caller is granted, by every word of a query, newest first, no body |
|
||||
| `mail_read` | one message's metadata and text, or one attachment's bytes, to a caller granted its account |
|
||||
| `mail_health` | healthy; unhealthy naming the account and why (connection down, or no answer for five minutes); waits naming `mail-credential-<account>` while its password is not given |
|
||||
|
||||
The caller is the runtime's word, never an argument.
|
||||
|
||||
## Its processes
|
||||
|
||||
`prepare` brings the index's schema to this version's (numbered migrations, and `schema.sql` for a fresh database).
|
||||
`keep` watches the accounts and is judged by its unit. `watch` judges every minute and is judged by `mail_health`.
|
||||
@@ -0,0 +1,260 @@
|
||||
package main
|
||||
|
||||
// The operator's mail accounts and who may read them (novox/hq ADR 0359 §1, §4, §5): two trusted settings, each a
|
||||
// file the mesh writes from the module's settings and changes only on the operator's approval. `accounts` says which
|
||||
// server each account is on and how to sign in (never a credential); `grants` says, per account, which callers may
|
||||
// read its mail. The password of an account is the module's own secret mail-credential-<account>, given at the desk
|
||||
// (ADR 0283 §6), at the path the secrets directory and that name make.
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"regexp"
|
||||
"sort"
|
||||
"strings"
|
||||
)
|
||||
|
||||
// The kinds of account an adapter exists for. Only `imap` is built (ADR 0359 §5); `imap-oauth` and `graph` are later
|
||||
// decisions (§7), and an account of a kind not listed here is refused by name.
|
||||
const KindIMAP = "imap"
|
||||
|
||||
// Kinds is every kind this version has an adapter for.
|
||||
var Kinds = []string{KindIMAP}
|
||||
|
||||
// SecretFamily is the prefix of every account's own secret: mail-credential-<account> (ADR 0283 §6).
|
||||
const SecretFamily = "mail-credential-"
|
||||
|
||||
// accountName is what an account may be called: a member name of the secret family (ADR 0283 §6) and one token of a
|
||||
// bus subject, so never a dot, a slash or an upper-case letter.
|
||||
var accountName = regexp.MustCompile(`^[a-z0-9][a-z0-9-]{0,39}$`)
|
||||
|
||||
// Server is where an account's mail is reached.
|
||||
type Server struct {
|
||||
Host string `json:"host"`
|
||||
Port int `json:"port"`
|
||||
// Security is tls (TLS from the first byte) or starttls (plain, then upgraded before signing in). Nothing
|
||||
// signs in over a connection that is not encrypted.
|
||||
Security string `json:"security"`
|
||||
}
|
||||
|
||||
// Address is host:port.
|
||||
func (s Server) Address() string { return fmt.Sprintf("%s:%d", s.Host, s.Port) }
|
||||
|
||||
// Account is one of the operator's mail accounts, as the setting `accounts` declares it.
|
||||
type Account struct {
|
||||
// Name is the key the account has in the setting; set when the setting is read.
|
||||
Name string `json:"-"`
|
||||
Kind string `json:"kind"`
|
||||
// Address is the account's mail address, said only to a caller granted the account.
|
||||
Address string `json:"address"`
|
||||
// User is the name it signs in with; the address when empty.
|
||||
User string `json:"user,omitempty"`
|
||||
Incoming Server `json:"incoming"`
|
||||
Outgoing *Server `json:"outgoing,omitempty"`
|
||||
// Folders are the folders watched besides the inbox; every folder the server lists when empty.
|
||||
Folders []string `json:"folders,omitempty"`
|
||||
}
|
||||
|
||||
// SignInAs is the name the account signs in with.
|
||||
func (a Account) SignInAs() string {
|
||||
if a.User != "" {
|
||||
return a.User
|
||||
}
|
||||
return a.Address
|
||||
}
|
||||
|
||||
// Secret is the name of the account's own secret.
|
||||
func (a Account) Secret() string { return SecretFamily + a.Name }
|
||||
|
||||
// AccountsFile is the shape of the trusted setting `accounts`, as the mesh writes it.
|
||||
type AccountsFile struct {
|
||||
Accounts map[string]Account `json:"accounts"`
|
||||
}
|
||||
|
||||
// ReadAccounts reads the accounts setting, refusing it whole when any account in it cannot be watched: an account
|
||||
// half-understood would be watched wrongly, and the operator approved the setting as one.
|
||||
func ReadAccounts(path string) ([]Account, error) {
|
||||
if path == "" {
|
||||
return nil, errors.New("the module was started without its accounts setting (MESH_MAIL_ACCOUNTS)")
|
||||
}
|
||||
raw, err := os.ReadFile(path)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("the accounts setting cannot be read: %w", err)
|
||||
}
|
||||
return ParseAccounts(raw)
|
||||
}
|
||||
|
||||
// ParseAccounts reads the accounts setting's content.
|
||||
func ParseAccounts(raw []byte) ([]Account, error) {
|
||||
var f AccountsFile
|
||||
dec := json.NewDecoder(strings.NewReader(string(raw)))
|
||||
dec.DisallowUnknownFields()
|
||||
if err := dec.Decode(&f); err != nil {
|
||||
return nil, fmt.Errorf("the accounts setting is not the module's: %v", err)
|
||||
}
|
||||
var out []Account
|
||||
var problems []string
|
||||
for name, a := range f.Accounts {
|
||||
a.Name = name
|
||||
if err := a.Check(); err != nil {
|
||||
problems = append(problems, err.Error())
|
||||
continue
|
||||
}
|
||||
out = append(out, a)
|
||||
}
|
||||
if len(problems) > 0 {
|
||||
sort.Strings(problems)
|
||||
return nil, errors.New("the accounts setting is refused: " + strings.Join(problems, "; "))
|
||||
}
|
||||
sort.Slice(out, func(i, j int) bool { return out[i].Name < out[j].Name })
|
||||
return out, nil
|
||||
}
|
||||
|
||||
// Check says why an account cannot be watched, or nil.
|
||||
func (a Account) Check() error {
|
||||
say := func(format string, args ...any) error {
|
||||
return fmt.Errorf("the account %q: %s", a.Name, fmt.Sprintf(format, args...))
|
||||
}
|
||||
if !accountName.MatchString(a.Name) {
|
||||
return say("is not an account's name: a lower-case letter or digit, then lower-case letters, digits and hyphens, at most 40")
|
||||
}
|
||||
known := false
|
||||
for _, k := range Kinds {
|
||||
known = known || a.Kind == k
|
||||
}
|
||||
if !known {
|
||||
return say("is of the kind %q, and this version watches only %s", a.Kind, strings.Join(Kinds, ", "))
|
||||
}
|
||||
if !strings.Contains(a.Address, "@") {
|
||||
return say("has no mail address")
|
||||
}
|
||||
if err := a.Incoming.check("incoming"); err != nil {
|
||||
return say("%v", err)
|
||||
}
|
||||
if a.Outgoing != nil {
|
||||
if err := a.Outgoing.check("outgoing"); err != nil {
|
||||
return say("%v", err)
|
||||
}
|
||||
}
|
||||
for _, f := range a.Folders {
|
||||
if strings.TrimSpace(f) == "" || strings.EqualFold(f, "INBOX") {
|
||||
return say("lists the folder %q, and the inbox is always watched", f)
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s Server) check(which string) error {
|
||||
switch {
|
||||
case strings.TrimSpace(s.Host) == "" || strings.ContainsAny(s.Host, " /:"):
|
||||
return fmt.Errorf("its %s server has no host name", which)
|
||||
case s.Port <= 0 || s.Port > 65535:
|
||||
return fmt.Errorf("its %s server's port %d is not a port", which, s.Port)
|
||||
case s.Security != "tls" && s.Security != "starttls":
|
||||
return fmt.Errorf("its %s server's security is %q, and only tls or starttls signs in", which, s.Security)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// SecretPath is where the account's own secret is placed in the secrets directory, as the manifest declares the
|
||||
// family: <dir>/mail-credential-<account>.secret.
|
||||
func SecretPath(dir string, a Account) string { return filepath.Join(dir, a.Secret()+".secret") }
|
||||
|
||||
// ReadSecret is the account's password, or given false when nobody gave it yet: the mesh never makes a member of the
|
||||
// family, so a member not given is no file (ADR 0283 §6).
|
||||
func ReadSecret(dir string, a Account) (string, bool, error) {
|
||||
raw, err := os.ReadFile(SecretPath(dir, a))
|
||||
if errors.Is(err, os.ErrNotExist) {
|
||||
return "", false, nil
|
||||
}
|
||||
if err != nil {
|
||||
return "", false, err
|
||||
}
|
||||
pw := strings.TrimRight(string(raw), "\r\n")
|
||||
if pw == "" {
|
||||
return "", false, nil
|
||||
}
|
||||
return pw, true, nil
|
||||
}
|
||||
|
||||
// SecretGiven says whether the account's own secret was given.
|
||||
//
|
||||
// It looks at the file and never reads it: whoever judges whether a password was given has no need of the password.
|
||||
// A file that is there and not empty was given; whether it signs in is the keeper's to say.
|
||||
func SecretGiven(dir string, a Account) bool {
|
||||
info, err := os.Stat(SecretPath(dir, a))
|
||||
return err == nil && info.Mode().IsRegular() && info.Size() > 0
|
||||
}
|
||||
|
||||
// ---- grants ------------------------------------------------------------------------------------------------------
|
||||
|
||||
// Grants is the trusted setting `grants`: per account, the callers that may read its mail (ADR 0359 §4). An entry is a
|
||||
// module's name (granted wherever it runs), a module on one machine (`<machine>.<module>`) or a person
|
||||
// (`person.<name>`), as the runtime names callers.
|
||||
type Grants map[string][]string
|
||||
|
||||
// GrantsFile is the shape of the setting as the mesh writes it.
|
||||
type GrantsFile struct {
|
||||
Grants Grants `json:"grants"`
|
||||
}
|
||||
|
||||
// ReadGrants reads the grants setting. A setting that cannot be read grants nothing, and says so.
|
||||
func ReadGrants(path string) (Grants, error) {
|
||||
if path == "" {
|
||||
return Grants{}, errors.New("the module was started without its grants setting (MESH_MAIL_GRANTS)")
|
||||
}
|
||||
raw, err := os.ReadFile(path)
|
||||
if err != nil {
|
||||
return Grants{}, fmt.Errorf("the grants setting cannot be read: %w", err)
|
||||
}
|
||||
var f GrantsFile
|
||||
dec := json.NewDecoder(strings.NewReader(string(raw)))
|
||||
dec.DisallowUnknownFields()
|
||||
if err := dec.Decode(&f); err != nil {
|
||||
return Grants{}, fmt.Errorf("the grants setting is not the module's: %v", err)
|
||||
}
|
||||
if f.Grants == nil {
|
||||
f.Grants = Grants{}
|
||||
}
|
||||
return f.Grants, nil
|
||||
}
|
||||
|
||||
// Unproven is the caller the runtime proved nobody to be; it is granted nothing.
|
||||
const Unproven = "unproven"
|
||||
|
||||
// Allows says whether the caller may read the account's mail. Nobody is granted an account by default, an unproven
|
||||
// caller never, and a wildcard is no entry.
|
||||
func (g Grants) Allows(caller, account string) bool {
|
||||
if caller == "" || caller == Unproven {
|
||||
return false
|
||||
}
|
||||
for _, entry := range g[account] {
|
||||
if entry == "" || strings.ContainsAny(entry, "*>") {
|
||||
continue
|
||||
}
|
||||
if entry == caller {
|
||||
return true
|
||||
}
|
||||
// A module's name alone grants it on every machine: the caller is <machine>.<module>.
|
||||
if !strings.Contains(entry, ".") {
|
||||
if _, module, ok := strings.Cut(caller, "."); ok && module == entry && !strings.HasPrefix(caller, "person.") {
|
||||
return true
|
||||
}
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
// Granted is the accounts, of those given, the caller may read.
|
||||
func (g Grants) Granted(caller string, accounts []Account) []string {
|
||||
var out []string
|
||||
for _, a := range accounts {
|
||||
if g.Allows(caller, a.Name) {
|
||||
out = append(out, a.Name)
|
||||
}
|
||||
}
|
||||
return out
|
||||
}
|
||||
@@ -0,0 +1,124 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"os"
|
||||
"path/filepath"
|
||||
"reflect"
|
||||
"strings"
|
||||
"testing"
|
||||
)
|
||||
|
||||
// The first stage's account, as the operator sets the setting `accounts` (ADR 0359 §5): its address and its server,
|
||||
// never a credential; its password is the own secret mail-credential-mesh.
|
||||
const stageOne = `{"accounts": {"mesh": {"kind": "imap", "address": "jschoubben@novox.be",
|
||||
"incoming": {"host": "mail.novox.be", "port": 993, "security": "tls"}}}}`
|
||||
|
||||
func TestTheFirstStagesAccountIsReadAndItsSecretNamed(t *testing.T) {
|
||||
got, err := ParseAccounts([]byte(stageOne))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if len(got) != 1 || got[0].Name != "mesh" || got[0].Kind != KindIMAP || got[0].SignInAs() != "jschoubben@novox.be" ||
|
||||
got[0].Incoming.Address() != "mail.novox.be:993" || got[0].Secret() != "mail-credential-mesh" || got[0].Outgoing != nil {
|
||||
t.Fatalf("%+v", got)
|
||||
}
|
||||
if p := SecretPath("/state", got[0]); p != "/state/mail-credential-mesh.secret" {
|
||||
t.Fatalf("the secret is placed at %s", p)
|
||||
}
|
||||
}
|
||||
|
||||
// A second account of the kind imap on another server — Gmail with an app password, another member of the family —
|
||||
// is configuration only: no code changes for it. Each account may also say its outgoing server, which this stage
|
||||
// reads and never uses (it sends nothing).
|
||||
func TestASecondIMAPAccountIsConfigurationOnly(t *testing.T) {
|
||||
got, err := ParseAccounts([]byte(`{"accounts": {
|
||||
"mesh": {"kind": "imap", "address": "jschoubben@novox.be", "incoming": {"host": "mail.novox.be", "port": 993, "security": "tls"}},
|
||||
"gmail": {"kind": "imap", "address": "someone@gmail.com", "incoming": {"host": "imap.gmail.com", "port": 993, "security": "tls"},
|
||||
"outgoing": {"host": "smtp.gmail.com", "port": 465, "security": "tls"}, "folders": ["[Gmail]/Sent Mail"]}}}`))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if len(got) != 2 || got[0].Name != "gmail" || got[0].Secret() != "mail-credential-gmail" || got[0].Outgoing.Address() != "smtp.gmail.com:465" ||
|
||||
!reflect.DeepEqual(got[0].Folders, []string{"[Gmail]/Sent Mail"}) {
|
||||
t.Fatalf("%+v", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestAnAccountThatCannotBeWatchedRefusesTheSetting(t *testing.T) {
|
||||
for name, raw := range map[string]string{
|
||||
"a later kind": `{"accounts": {"work": {"kind": "graph", "address": "a@b.c", "incoming": {"host": "h", "port": 1, "security": "tls"}}}}`,
|
||||
"no encryption": `{"accounts": {"x": {"kind": "imap", "address": "a@b.c", "incoming": {"host": "h", "port": 143, "security": "none"}}}}`,
|
||||
"a name with a dot": `{"accounts": {"my.box": {"kind": "imap", "address": "a@b.c", "incoming": {"host": "h", "port": 993, "security": "tls"}}}}`,
|
||||
"a password in it": `{"accounts": {"x": {"kind": "imap", "address": "a@b.c", "password": "p", "incoming": {"host": "h", "port": 993, "security": "tls"}}}}`,
|
||||
"no address": `{"accounts": {"x": {"kind": "imap", "incoming": {"host": "h", "port": 993, "security": "tls"}}}}`,
|
||||
"the inbox listed": `{"accounts": {"x": {"kind": "imap", "address": "a@b.c", "folders": ["INBOX"], "incoming": {"host": "h", "port": 993, "security": "tls"}}}}`,
|
||||
"a bad outgoing one": `{"accounts": {"x": {"kind": "imap", "address": "a@b.c", "incoming": {"host": "h", "port": 993, "security": "tls"}, "outgoing": {"host": "", "port": 465, "security": "tls"}}}}`,
|
||||
} {
|
||||
if _, err := ParseAccounts([]byte(raw)); err == nil {
|
||||
t.Errorf("%s: accepted", name)
|
||||
}
|
||||
}
|
||||
if got, err := ParseAccounts([]byte(`{"accounts": {}}`)); err != nil || len(got) != 0 {
|
||||
t.Fatalf("the manifest's default: %v %v", got, err)
|
||||
}
|
||||
}
|
||||
|
||||
// Grants (ADR 0359 §4): nobody by default; a module's name grants it wherever it runs; a module on one machine, or a
|
||||
// person, by its full name; never the unproven, never by a wildcard, never one account for another.
|
||||
func TestGrantsAreByAccountAndByProvenCaller(t *testing.T) {
|
||||
g := Grants{"mesh": {"triage", "novox.desk", "person.jochen", "*"}}
|
||||
for caller, want := range map[string]bool{
|
||||
"novox.triage": true, "shanks.triage": true, "novox.desk": true, "g14.desk": false, "person.jochen": true,
|
||||
"person.triage": false, "novox.other": false, Unproven: false, "": false, "novox.*": false,
|
||||
} {
|
||||
if got := g.Allows(caller, "mesh"); got != want {
|
||||
t.Errorf("%s: %v", caller, got)
|
||||
}
|
||||
}
|
||||
if g.Allows("novox.triage", "gmail") || (Grants{}).Allows("novox.triage", "mesh") {
|
||||
t.Fatal("a grant reached an account it does not name")
|
||||
}
|
||||
dir := t.TempDir()
|
||||
path := filepath.Join(dir, "grants.json")
|
||||
_ = os.WriteFile(path, []byte(`{"grants": {"mesh": ["triage"]}}`), 0o600)
|
||||
read, err := ReadGrants(path)
|
||||
if err != nil || !read.Allows("novox.triage", "mesh") {
|
||||
t.Fatalf("%v %v", read, err)
|
||||
}
|
||||
_ = os.WriteFile(path, []byte(`{"grants": {"mesh": "triage"}}`), 0o600)
|
||||
if read, err := ReadGrants(path); err == nil || len(read) != 0 {
|
||||
t.Fatal("a grants setting it cannot read granted something")
|
||||
}
|
||||
}
|
||||
|
||||
// Whether a password was given is read from the file's presence and size alone (issue 733): an empty file is not
|
||||
// given, a file the judge cannot read is still given, and the password is never read to say so.
|
||||
func TestASecretIsGivenByItsFileNotByReadingIt(t *testing.T) {
|
||||
dir := t.TempDir()
|
||||
a := testAccount()
|
||||
if SecretGiven(dir, a) {
|
||||
t.Fatal("given before anything was written")
|
||||
}
|
||||
_ = os.WriteFile(SecretPath(dir, a), nil, 0o600)
|
||||
if SecretGiven(dir, a) {
|
||||
t.Fatal("an empty file is a password")
|
||||
}
|
||||
_ = os.WriteFile(SecretPath(dir, a), []byte("pw\n"), 0o600)
|
||||
if pw, given, _ := ReadSecret(dir, a); !given || pw != "pw" || strings.Contains(pw, "\n") {
|
||||
t.Fatalf("%q %v", pw, given)
|
||||
}
|
||||
// Unreadable to the judge, as a secret owned by root is to the watch: still given.
|
||||
if err := os.Chmod(SecretPath(dir, a), 0); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if !SecretGiven(dir, a) {
|
||||
t.Fatal("a password the judge cannot read is said not given")
|
||||
}
|
||||
src, _ := os.ReadFile("config.go")
|
||||
body := string(src)
|
||||
body = body[strings.Index(body, "func SecretGiven("):]
|
||||
body = body[:strings.Index(body, "\n}\n")]
|
||||
if strings.Contains(body, "ReadSecret") || strings.Contains(body, "ReadFile") || strings.Contains(body, "os.Open") {
|
||||
t.Fatalf("SecretGiven reads the password:\n%s", body)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,170 @@
|
||||
package main
|
||||
|
||||
// How mesh-mail is (novox/hq ADR 0359 §6). Its keeper's liveness is its unit's; its watching is judged here, by the
|
||||
// watch process, once a minute, and left in a status file the bundle's tool `mail_health` reads:
|
||||
//
|
||||
// - waits, naming the own secret mail-credential-<account>, for an account whose password nobody gave yet (ADR 0283),
|
||||
// so the condition is needs-operator and the gate passes;
|
||||
// - unhealthy, naming the account and why, when its connection is down or its server has not answered for
|
||||
// SilentAfter (five minutes) — an IDLE that went silent;
|
||||
// - healthy otherwise.
|
||||
//
|
||||
// A fault beside a wait is unhealthy and says both.
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"os"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/health"
|
||||
)
|
||||
|
||||
// HealthTool is the module's tool the node-engine asks how it is.
|
||||
const HealthTool = "mail_health"
|
||||
|
||||
// WatchEvery is how often the watch judges; StatusStale how old its verdict may be before the tool says it stopped.
|
||||
const (
|
||||
WatchEvery = time.Minute
|
||||
StatusStale = 3 * WatchEvery
|
||||
// SignInGrace is how long a password just given may wait for its keeper to sign in before that is a fault.
|
||||
SignInGrace = 3 * time.Minute
|
||||
)
|
||||
|
||||
// Judge is the module's health from what the watch reads: the accounts setting (or why it cannot be read), whether the
|
||||
// index serves, which passwords were given, and each account's state as its keeper said it.
|
||||
func Judge(accounts []Account, accountsErr, readyErr error, given func(Account) bool, states map[string]AccountState, now time.Time) health.Answer {
|
||||
if accountsErr != nil {
|
||||
return health.Unhealthy(accountsErr.Error())
|
||||
}
|
||||
var problems, fine []string
|
||||
var waits []health.Wait
|
||||
if readyErr != nil {
|
||||
problems = append(problems, "the index cannot serve: "+readyErr.Error())
|
||||
}
|
||||
for _, a := range accounts {
|
||||
if !given(a) {
|
||||
waits = append(waits, health.Wait{
|
||||
Part: "the mail account " + a.Name,
|
||||
Secret: a.Secret(),
|
||||
What: "the password of the mail account " + a.Name,
|
||||
})
|
||||
continue
|
||||
}
|
||||
if readyErr != nil {
|
||||
continue
|
||||
}
|
||||
st, said := states[a.Name]
|
||||
switch {
|
||||
case !said:
|
||||
problems = append(problems, fmt.Sprintf("the account %s: its keeper has said nothing of it", a.Name))
|
||||
case st.State == AccountWaits:
|
||||
if now.Sub(st.Said) > SignInGrace {
|
||||
problems = append(problems, fmt.Sprintf("the account %s: its password was given, and its keeper has not signed in for %s",
|
||||
a.Name, now.Sub(st.Said).Round(time.Second)))
|
||||
} else {
|
||||
fine = append(fine, a.Name+" is signing in")
|
||||
}
|
||||
case st.State == AccountFailing:
|
||||
problems = append(problems, fmt.Sprintf("the account %s is not watched: %s", a.Name, st.Why))
|
||||
case st.Heard.IsZero():
|
||||
problems = append(problems, fmt.Sprintf("the account %s: its server has not answered since its keeper started", a.Name))
|
||||
case now.Sub(st.Heard) > SilentAfter:
|
||||
problems = append(problems, fmt.Sprintf("the account %s: its server has not answered for %s (its IDLE on the inbox is silent)",
|
||||
a.Name, now.Sub(st.Heard).Round(time.Second)))
|
||||
default:
|
||||
fine = append(fine, fmt.Sprintf("%s is %s (%s)", a.Name, st.State, st.Why))
|
||||
}
|
||||
}
|
||||
waitSaid := ""
|
||||
if len(waits) > 0 {
|
||||
var names []string
|
||||
for _, w := range waits {
|
||||
names = append(names, w.Secret)
|
||||
}
|
||||
waitSaid = "waits for its password: " + strings.Join(names, ", ")
|
||||
}
|
||||
switch {
|
||||
case len(problems) > 0:
|
||||
why := strings.Join(problems, "; ")
|
||||
if waitSaid != "" {
|
||||
why += "; and it " + waitSaid
|
||||
}
|
||||
return health.Unhealthy(why, waits...)
|
||||
case len(waits) > 0:
|
||||
why := "nothing is wrong, and it " + waitSaid
|
||||
if len(fine) > 0 {
|
||||
why += "; " + strings.Join(fine, "; ")
|
||||
}
|
||||
return health.Waiting(why, waits...)
|
||||
case len(accounts) == 0:
|
||||
return health.Healthy("no mail account is declared, so nothing is watched")
|
||||
}
|
||||
return health.Healthy(fmt.Sprintf("%d account(s) watched: %s", len(accounts), strings.Join(fine, "; ")))
|
||||
}
|
||||
|
||||
// Status is the watch's verdict as it leaves it: the answer, and when it looked.
|
||||
type Status struct {
|
||||
health.Answer
|
||||
At time.Time `json:"at"`
|
||||
}
|
||||
|
||||
func writeStatus(path string, s Status) error {
|
||||
raw, err := json.Marshal(s)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
tmp := path + ".next"
|
||||
if err := os.WriteFile(tmp, raw, 0o600); err != nil {
|
||||
return err
|
||||
}
|
||||
return os.Rename(tmp, path)
|
||||
}
|
||||
|
||||
// ReadStatus is the watch's last verdict, as the tool answers it: unhealthy when there is none, when it cannot be read,
|
||||
// or when it is older than StatusStale.
|
||||
func ReadStatus(path string, now time.Time) health.Answer {
|
||||
raw, err := os.ReadFile(path)
|
||||
if err != nil {
|
||||
return health.Unhealthy("the watch has left no verdict yet: " + err.Error())
|
||||
}
|
||||
var s Status
|
||||
if err := json.Unmarshal(raw, &s); err != nil {
|
||||
return health.Unhealthy("the watch's verdict cannot be read")
|
||||
}
|
||||
if age := now.Sub(s.At); age > StatusStale {
|
||||
return health.Unhealthy(fmt.Sprintf("the watch has not looked for %s (its last verdict: %s)", age.Round(time.Second), s.Why))
|
||||
}
|
||||
// Read through the SDK's reader, so a wait it cannot check is never excused.
|
||||
a, err := health.Read(mustJSON(s.Answer))
|
||||
if err != nil {
|
||||
return health.Unhealthy("the watch's verdict cannot be read: " + err.Error())
|
||||
}
|
||||
return a
|
||||
}
|
||||
|
||||
func mustJSON(v any) []byte {
|
||||
raw, _ := json.Marshal(v)
|
||||
return raw
|
||||
}
|
||||
|
||||
// watchOnce is one look: what the watch reads, judged.
|
||||
func watchOnce(ctx context.Context, env Env, open func(context.Context) (Store, func(), error), now time.Time) health.Answer {
|
||||
accounts, accountsErr := ReadAccounts(env.Accounts)
|
||||
var readyErr error
|
||||
states := map[string]AccountState{}
|
||||
st, done, err := open(ctx)
|
||||
if err != nil {
|
||||
readyErr = err
|
||||
} else {
|
||||
defer done()
|
||||
if readyErr = st.Ready(ctx); readyErr == nil {
|
||||
if states, err = st.States(ctx); err != nil {
|
||||
readyErr = err
|
||||
}
|
||||
}
|
||||
}
|
||||
return Judge(accounts, accountsErr, readyErr, func(a Account) bool { return SecretGiven(env.Secrets, a) }, states, now)
|
||||
}
|
||||
@@ -0,0 +1,116 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/health"
|
||||
)
|
||||
|
||||
var judgedAt = time.Date(2026, 10, 14, 18, 0, 0, 0, time.UTC)
|
||||
|
||||
func given(names ...string) func(Account) bool {
|
||||
return func(a Account) bool { return contains(names, a.Name) }
|
||||
}
|
||||
|
||||
// Before the password is given, mail_health answers waits, naming the secret, which the node-engine reads as a wait
|
||||
// the controller can excuse (ADR 0283 §1, §3): the family mail-credential-* is declared issued outside the mesh.
|
||||
func TestItWaitsNamingTheSecretUntilThePasswordIsGiven(t *testing.T) {
|
||||
a := Judge([]Account{testAccount()}, nil, nil, given(), nil, judgedAt)
|
||||
if a.State != health.StateWaits || len(a.Waits) != 1 || a.Waits[0].Secret != "mail-credential-mesh" || a.Healthy {
|
||||
t.Fatalf("%+v", a)
|
||||
}
|
||||
read, err := health.Read(mustJSON(a))
|
||||
if err != nil || read.State != health.StateWaits {
|
||||
t.Fatalf("the node-engine reads it as %+v (%v)", read, err)
|
||||
}
|
||||
if strings.Contains(a.Waits[0].What, "@") || strings.Contains(a.Why, "@") {
|
||||
t.Fatalf("the wait says an address: %+v", a)
|
||||
}
|
||||
}
|
||||
|
||||
// Unhealthy, naming the account and why, when its connection is down or its server silent for five minutes.
|
||||
func TestItIsUnhealthyWhenAnAccountIsDownOrSilent(t *testing.T) {
|
||||
acc := []Account{testAccount()}
|
||||
down := map[string]AccountState{"mesh": {Account: "mesh", State: AccountFailing, Why: "its server mail.example cannot be reached", Said: judgedAt}}
|
||||
if a := Judge(acc, nil, nil, given("mesh"), down, judgedAt); a.State != health.StateUnhealthy || !strings.Contains(a.Why, "the account mesh is not watched: its server") {
|
||||
t.Fatalf("down: %+v", a)
|
||||
}
|
||||
silent := map[string]AccountState{"mesh": {Account: "mesh", State: AccountCurrent, Heard: judgedAt.Add(-SilentAfter - time.Second), Said: judgedAt}}
|
||||
if a := Judge(acc, nil, nil, given("mesh"), silent, judgedAt); a.State != health.StateUnhealthy || !strings.Contains(a.Why, "IDLE on the inbox is silent") {
|
||||
t.Fatalf("silent: %+v", a)
|
||||
}
|
||||
fine := map[string]AccountState{"mesh": {Account: "mesh", State: AccountCurrent, Why: "watching", Heard: judgedAt.Add(-time.Minute), Said: judgedAt}}
|
||||
if a := Judge(acc, nil, nil, given("mesh"), fine, judgedAt); a.State != health.StateHealthy || !a.Healthy {
|
||||
t.Fatalf("fine: %+v", a)
|
||||
}
|
||||
if a := Judge(acc, nil, errors.New("the schema is behind"), given("mesh"), fine, judgedAt); a.State != health.StateUnhealthy {
|
||||
t.Fatalf("an index that cannot serve: %+v", a)
|
||||
}
|
||||
if a := Judge(nil, errors.New("the accounts setting is refused"), nil, given(), nil, judgedAt); a.State != health.StateUnhealthy {
|
||||
t.Fatalf("a setting it cannot read: %+v", a)
|
||||
}
|
||||
// A fault beside a wait is unhealthy and says both (ADR 0283 §1).
|
||||
two := []Account{testAccount(), {Name: "gmail", Kind: KindIMAP, Address: "x@example.com", Incoming: Server{Host: "h", Port: 993, Security: "tls"}}}
|
||||
a := Judge(two, nil, nil, given("mesh"), down, judgedAt)
|
||||
if a.State != health.StateUnhealthy || len(a.Waits) != 1 || a.Waits[0].Secret != "mail-credential-gmail" || !strings.Contains(a.Why, "mail-credential-gmail") {
|
||||
t.Fatalf("both: %+v", a)
|
||||
}
|
||||
// A password just given is signing in for a while; not for ever.
|
||||
signing := map[string]AccountState{"mesh": {Account: "mesh", State: AccountWaits, Said: judgedAt.Add(-time.Minute)}}
|
||||
if a := Judge(acc, nil, nil, given("mesh"), signing, judgedAt); a.State != health.StateHealthy {
|
||||
t.Fatalf("signing in: %+v", a)
|
||||
}
|
||||
signing["mesh"] = AccountState{Account: "mesh", State: AccountWaits, Said: judgedAt.Add(-SignInGrace - time.Minute)}
|
||||
if a := Judge(acc, nil, nil, given("mesh"), signing, judgedAt); a.State != health.StateUnhealthy {
|
||||
t.Fatalf("never signed in: %+v", a)
|
||||
}
|
||||
if a := Judge(nil, nil, nil, given(), nil, judgedAt); a.State != health.StateHealthy || !strings.Contains(a.Why, "nothing is watched") {
|
||||
t.Fatalf("no account: %+v", a)
|
||||
}
|
||||
}
|
||||
|
||||
// The tool reads the watch's verdict, and says when the watch stopped looking.
|
||||
func TestTheHealthToolReadsTheWatchsVerdictAndSaysWhenItIsStale(t *testing.T) {
|
||||
path := filepath.Join(t.TempDir(), "health.json")
|
||||
if a := ReadStatus(path, judgedAt); a.State != health.StateUnhealthy {
|
||||
t.Fatalf("no verdict: %+v", a)
|
||||
}
|
||||
waits := Judge([]Account{testAccount()}, nil, nil, given(), nil, judgedAt)
|
||||
if err := writeStatus(path, Status{Answer: waits, At: judgedAt}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if a := ReadStatus(path, judgedAt.Add(time.Minute)); a.State != health.StateWaits || a.Waits[0].Secret != "mail-credential-mesh" {
|
||||
t.Fatalf("%+v", a)
|
||||
}
|
||||
if a := ReadStatus(path, judgedAt.Add(StatusStale+time.Second)); a.State != health.StateUnhealthy || !strings.Contains(a.Why, "has not looked") {
|
||||
t.Fatalf("stale: %+v", a)
|
||||
}
|
||||
}
|
||||
|
||||
// One look of the watch, end to end over the files the mesh writes: the accounts setting, the secrets directory and
|
||||
// the index.
|
||||
func TestTheWatchJudgesFromTheFilesAndTheIndex(t *testing.T) {
|
||||
dir := t.TempDir()
|
||||
env := Env{Accounts: filepath.Join(dir, "accounts.json"), Secrets: dir}
|
||||
writeFile(t, env.Accounts, stageOne)
|
||||
st := newMemStore()
|
||||
open := func(context.Context) (Store, func(), error) { return st, func() {}, nil }
|
||||
if a := watchOnce(context.Background(), env, open, judgedAt); a.State != health.StateWaits {
|
||||
t.Fatalf("%+v", a)
|
||||
}
|
||||
giveSecret(t, dir, Account{Name: "mesh"}, "pw")
|
||||
_ = st.SetState(context.Background(), AccountState{Account: "mesh", State: AccountCurrent, Heard: judgedAt, Said: judgedAt})
|
||||
if a := watchOnce(context.Background(), env, open, judgedAt); a.State != health.StateHealthy {
|
||||
t.Fatalf("%+v", a)
|
||||
}
|
||||
raw, _ := json.Marshal(Status{Answer: health.Healthy("x"), At: judgedAt})
|
||||
if !strings.Contains(string(raw), `"state":"healthy"`) || !strings.Contains(string(raw), `"at":`) {
|
||||
t.Fatalf("the status file's shape: %s", raw)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,576 @@
|
||||
package main
|
||||
|
||||
// The adapter for an account of the kind `imap` (novox/hq ADR 0359 §1, §2, §5). For each account it keeps one
|
||||
// connection: IDLE on the inbox, and every PollEvery a look at every other folder. Each look compares what the server
|
||||
// holds with the index and says each difference as an event: a message added, its flags changed, or removed. The first
|
||||
// look at a folder indexes what is there without saying it — what the bus is told is what changed since the mesh
|
||||
// started watching, never the history of a mailbox.
|
||||
//
|
||||
// **It only reads.** Every folder is opened read-only (EXAMINE), and every message is fetched with BODY.PEEK, so
|
||||
// watching never marks a message seen. The client is used through reader alone, which has no command that flags,
|
||||
// moves, copies, deletes, appends or sends (§5); a test holds the code to that.
|
||||
|
||||
import (
|
||||
"context"
|
||||
"crypto/tls"
|
||||
"errors"
|
||||
"fmt"
|
||||
"sort"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/emersion/go-imap/v2"
|
||||
"github.com/emersion/go-imap/v2/imapclient"
|
||||
)
|
||||
|
||||
// How the keeper paces itself. IdleCycle is how long one IDLE lasts before it is ended and begun again, which proves
|
||||
// the server still answers; SilentAfter is how long without an answer makes the account unhealthy (§6).
|
||||
const (
|
||||
IdleCycle = 2 * time.Minute
|
||||
PollEvery = 5 * time.Minute
|
||||
SilentAfter = 5 * time.Minute
|
||||
RetryFirst = 30 * time.Second
|
||||
RetryMost = 5 * time.Minute
|
||||
FetchBatch = 25
|
||||
// BatchBytes bounds the mail one fetch of bodies holds; SizeBatch is how many sizes one fetch asks.
|
||||
BatchBytes = 64 << 20
|
||||
SizeBatch = 500
|
||||
)
|
||||
|
||||
// Batches cuts the uids, in order, into batches of at most most messages whose sizes add up to at most limit bytes. A
|
||||
// message larger than limit is a batch of its own. A size not known counts as nothing: the server listed the message,
|
||||
// and the fetch says what it holds.
|
||||
func Batches(uids []uint32, sizes map[uint32]int64, limit int64, most int) [][]uint32 {
|
||||
var out [][]uint32
|
||||
var cur []uint32
|
||||
var bytes int64
|
||||
for _, u := range uids {
|
||||
size := sizes[u]
|
||||
if len(cur) > 0 && (len(cur) >= most || bytes+size > limit) {
|
||||
out = append(out, cur)
|
||||
cur, bytes = nil, 0
|
||||
}
|
||||
cur = append(cur, u)
|
||||
bytes += size
|
||||
}
|
||||
if len(cur) > 0 {
|
||||
out = append(out, cur)
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
func uidSet(uids []uint32) imap.UIDSet {
|
||||
set := make([]imap.UID, len(uids))
|
||||
for i, u := range uids {
|
||||
set[i] = imap.UID(u)
|
||||
}
|
||||
return imap.UIDSetNum(set...)
|
||||
}
|
||||
|
||||
func orBytes(n, def int64) int64 {
|
||||
if n > 0 {
|
||||
return n
|
||||
}
|
||||
return def
|
||||
}
|
||||
|
||||
// reader is the part of an IMAP client the keeper uses: nothing in it changes a mailbox.
|
||||
type reader interface {
|
||||
Login(username, password string) *imapclient.Command
|
||||
List(ref, pattern string, options *imap.ListOptions) *imapclient.ListCommand
|
||||
Select(mailbox string, options *imap.SelectOptions) *imapclient.SelectCommand
|
||||
Fetch(numSet imap.NumSet, options *imap.FetchOptions) *imapclient.FetchCommand
|
||||
Idle() (*imapclient.IdleCommand, error)
|
||||
Caps() imap.CapSet
|
||||
Closed() <-chan struct{}
|
||||
Logout() *imapclient.Command
|
||||
Close() error
|
||||
}
|
||||
|
||||
// Dialer opens a connection to an account's server, handing what the server says unasked to the handler.
|
||||
type Dialer func(a Account, handler *imapclient.UnilateralDataHandler) (reader, error)
|
||||
|
||||
// DialServer is the dialer for the mesh: TLS from the first byte, or STARTTLS before signing in, as the account says.
|
||||
func DialServer(a Account, handler *imapclient.UnilateralDataHandler) (reader, error) {
|
||||
opts := &imapclient.Options{UnilateralDataHandler: handler, TLSConfig: &tls.Config{ServerName: a.Incoming.Host, MinVersion: tls.VersionTLS12}}
|
||||
var c *imapclient.Client
|
||||
var err error
|
||||
switch a.Incoming.Security {
|
||||
case "tls":
|
||||
c, err = imapclient.DialTLS(a.Incoming.Address(), opts)
|
||||
case "starttls":
|
||||
c, err = imapclient.DialStartTLS(a.Incoming.Address(), opts)
|
||||
default:
|
||||
err = fmt.Errorf("the account %s's security %q signs in over nothing this version speaks", a.Name, a.Incoming.Security)
|
||||
}
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return c, nil
|
||||
}
|
||||
|
||||
// Keeper watches the accounts and keeps the index.
|
||||
type Keeper struct {
|
||||
Store Store
|
||||
Objects Objects
|
||||
Dial Dialer
|
||||
SecretsDir string
|
||||
Logf func(format string, a ...any)
|
||||
Now func() time.Time
|
||||
|
||||
// Paces, for the tests; the constants above when zero.
|
||||
IdleCycle, PollEvery, RetryFirst time.Duration
|
||||
// BatchBytes, for the tests; the constant when zero.
|
||||
BatchBytes int64
|
||||
}
|
||||
|
||||
func (k *Keeper) now() time.Time {
|
||||
if k.Now != nil {
|
||||
return k.Now()
|
||||
}
|
||||
return time.Now()
|
||||
}
|
||||
|
||||
func (k *Keeper) logf(format string, a ...any) {
|
||||
if k.Logf != nil {
|
||||
k.Logf(format, a...)
|
||||
}
|
||||
}
|
||||
|
||||
func orDefault(d, def time.Duration) time.Duration {
|
||||
if d > 0 {
|
||||
return d
|
||||
}
|
||||
return def
|
||||
}
|
||||
|
||||
// Run watches every account until ctx ends.
|
||||
func (k *Keeper) Run(ctx context.Context, accounts []Account) {
|
||||
var wg sync.WaitGroup
|
||||
for _, a := range accounts {
|
||||
wg.Add(1)
|
||||
go func(a Account) {
|
||||
defer wg.Done()
|
||||
k.Watch(ctx, a)
|
||||
}(a)
|
||||
}
|
||||
wg.Wait()
|
||||
}
|
||||
|
||||
func (k *Keeper) say(ctx context.Context, a Account, state, why string, heard, synced time.Time) {
|
||||
if err := k.Store.SetState(ctx, AccountState{Account: a.Name, State: state, Why: why, Heard: heard, Synced: synced, Said: k.now()}); err != nil {
|
||||
k.logf("[mesh-mail] the account %s's state could not be kept: %v", a.Name, err)
|
||||
}
|
||||
}
|
||||
|
||||
// Watch keeps one account until ctx ends: waits while its secret is not given, signs in, watches, and on any fault
|
||||
// says it, waits a little longer each time, and begins again.
|
||||
func (k *Keeper) Watch(ctx context.Context, a Account) {
|
||||
retry := orDefault(k.RetryFirst, RetryFirst)
|
||||
var heard, synced time.Time
|
||||
for ctx.Err() == nil {
|
||||
pw, given, err := ReadSecret(k.SecretsDir, a)
|
||||
switch {
|
||||
case err != nil:
|
||||
k.say(ctx, a, AccountFailing, "its password was given and cannot be read: "+err.Error(), heard, synced)
|
||||
case !given:
|
||||
k.say(ctx, a, AccountWaits, "its password ("+a.Secret()+") has not been given", heard, synced)
|
||||
default:
|
||||
s := &session{k: k, a: a, wake: make(chan struct{}, 1), heard: heard, synced: synced}
|
||||
err = s.run(ctx, pw)
|
||||
heard, synced = s.heard, s.synced
|
||||
if ctx.Err() != nil {
|
||||
return
|
||||
}
|
||||
k.say(ctx, a, AccountFailing, err.Error(), heard, synced)
|
||||
k.logf("[mesh-mail] the account %s: %v; trying again in %s", a.Name, err, retry)
|
||||
}
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
case <-time.After(retry):
|
||||
}
|
||||
if given && err != nil {
|
||||
retry = min(retry*2, RetryMost)
|
||||
} else {
|
||||
retry = orDefault(k.RetryFirst, RetryFirst)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// session is one connection to one account's server.
|
||||
type session struct {
|
||||
k *Keeper
|
||||
a Account
|
||||
c reader
|
||||
wake chan struct{}
|
||||
heard time.Time
|
||||
synced time.Time
|
||||
// state and why are what the session last said of the account, and told when it said it.
|
||||
state, why string
|
||||
told time.Time
|
||||
}
|
||||
|
||||
// answered is the server having answered: kept, and told at least once a minute, so a long first look at a large
|
||||
// mailbox still shows the account alive to the health watch.
|
||||
func (s *session) answered() {
|
||||
s.heard = s.k.now()
|
||||
if s.state != "" && s.heard.Sub(s.told) >= time.Minute {
|
||||
s.tell(s.state, s.why)
|
||||
}
|
||||
}
|
||||
|
||||
// tell says the account's state.
|
||||
func (s *session) tell(state, why string) {
|
||||
s.state, s.why, s.told = state, why, s.k.now()
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
|
||||
defer cancel()
|
||||
s.k.say(ctx, s.a, state, why, s.heard, s.synced)
|
||||
}
|
||||
|
||||
func (s *session) poke() {
|
||||
select {
|
||||
case s.wake <- struct{}{}:
|
||||
default:
|
||||
}
|
||||
}
|
||||
|
||||
func (s *session) run(ctx context.Context, password string) error {
|
||||
handler := &imapclient.UnilateralDataHandler{
|
||||
Expunge: func(uint32) { s.poke() },
|
||||
Mailbox: func(*imapclient.UnilateralDataMailbox) { s.poke() },
|
||||
Fetch: func(*imapclient.FetchMessageData) { s.poke() },
|
||||
}
|
||||
c, err := s.k.Dial(s.a, handler)
|
||||
if err != nil {
|
||||
return fmt.Errorf("its server %s cannot be reached: %v", s.a.Incoming.Host, err)
|
||||
}
|
||||
s.c = c
|
||||
defer c.Close()
|
||||
if err := c.Login(s.a.SignInAs(), password).Wait(); err != nil {
|
||||
var said *imap.Error
|
||||
if errors.As(err, &said) {
|
||||
return fmt.Errorf("its server refused the sign-in (%s): its password may be wrong, and is given again at the desk", strings.TrimSpace(said.Text))
|
||||
}
|
||||
return fmt.Errorf("signing in failed: %v", err)
|
||||
}
|
||||
s.answered()
|
||||
s.tell(AccountSyncing, "signed in; looking at its folders")
|
||||
folders, err := s.folders()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
idles := c.Caps().Has(imap.CapIdle) || c.Caps().Has(imap.CapIMAP4rev2)
|
||||
var polled time.Time
|
||||
for ctx.Err() == nil {
|
||||
if polled.IsZero() || s.k.now().Sub(polled) >= orDefault(s.k.PollEvery, PollEvery) {
|
||||
for _, f := range folders {
|
||||
if err := s.k.syncFolder(ctx, c, s.a, f, s.answered); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
polled = s.k.now()
|
||||
}
|
||||
// The inbox last, so it is the folder selected while the session idles.
|
||||
if err := s.k.syncFolder(ctx, c, s.a, "INBOX", s.answered); err != nil {
|
||||
return err
|
||||
}
|
||||
s.synced = s.k.now()
|
||||
s.tell(AccountCurrent, fmt.Sprintf("watching the inbox and %d other folder(s)", len(folders)))
|
||||
if err := s.wait(ctx, idles); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
_ = c.Logout().Wait()
|
||||
return ctx.Err()
|
||||
}
|
||||
|
||||
// wait is one IDLE on the inbox, ended by what the server says unasked, by the cycle's end or by ctx. Ending it waits
|
||||
// for the server's answer, which is what proves the connection alive.
|
||||
func (s *session) wait(ctx context.Context, idles bool) error {
|
||||
cycle := time.NewTimer(orDefault(s.k.IdleCycle, IdleCycle))
|
||||
defer cycle.Stop()
|
||||
if !idles {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
case <-cycle.C:
|
||||
case <-s.c.Closed():
|
||||
return errors.New("its server closed the connection")
|
||||
}
|
||||
return nil
|
||||
}
|
||||
idle, err := s.c.Idle()
|
||||
if err != nil {
|
||||
return fmt.Errorf("its server refused to IDLE on the inbox: %v", err)
|
||||
}
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
case <-s.wake:
|
||||
case <-cycle.C:
|
||||
case <-s.c.Closed():
|
||||
}
|
||||
if err := idle.Close(); err != nil {
|
||||
return fmt.Errorf("its IDLE on the inbox ended badly: %v", err)
|
||||
}
|
||||
if err := idle.Wait(); err != nil {
|
||||
return fmt.Errorf("its IDLE on the inbox ended badly: %v", err)
|
||||
}
|
||||
s.answered()
|
||||
return nil
|
||||
}
|
||||
|
||||
// folders is what is watched besides the inbox: the account's list, or every folder the server can open.
|
||||
func (s *session) folders() ([]string, error) {
|
||||
if len(s.a.Folders) > 0 {
|
||||
return append([]string(nil), s.a.Folders...), nil
|
||||
}
|
||||
listed, err := s.c.List("", "*", nil).Collect()
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("its folders cannot be listed: %v", err)
|
||||
}
|
||||
s.answered()
|
||||
var out []string
|
||||
for _, l := range listed {
|
||||
if strings.EqualFold(l.Mailbox, "INBOX") || hasAttr(l.Attrs, imap.MailboxAttrNoSelect) || hasAttr(l.Attrs, imap.MailboxAttrNonExistent) {
|
||||
continue
|
||||
}
|
||||
out = append(out, l.Mailbox)
|
||||
}
|
||||
sort.Strings(out)
|
||||
return out, nil
|
||||
}
|
||||
|
||||
func hasAttr(attrs []imap.MailboxAttr, want imap.MailboxAttr) bool {
|
||||
for _, a := range attrs {
|
||||
if strings.EqualFold(string(a), string(want)) {
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
// syncFolder brings the index of one folder to what the server holds, and says each difference, unless this is the
|
||||
// first look at the folder.
|
||||
func (k *Keeper) syncFolder(ctx context.Context, c reader, a Account, folder string, answered func()) error {
|
||||
sel, err := c.Select(folder, &imap.SelectOptions{ReadOnly: true}).Wait()
|
||||
if err != nil {
|
||||
return fmt.Errorf("the folder %s cannot be opened: %v", folder, err)
|
||||
}
|
||||
answered()
|
||||
known, seen, err := k.Store.Folder(ctx, a.Name, folder)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if seen && known.UIDValidity != sel.UIDValidity {
|
||||
k.logf("[mesh-mail] the account %s's folder %s was renumbered by its server; it is indexed afresh", a.Name, folder)
|
||||
if err := k.Store.DropFolder(ctx, a.Name, folder); err != nil {
|
||||
return err
|
||||
}
|
||||
seen = false
|
||||
}
|
||||
server := map[uint32][]string{}
|
||||
if sel.NumMessages > 0 {
|
||||
all, err := c.Fetch(imap.UIDSet{{Start: 1, Stop: 0}}, &imap.FetchOptions{UID: true, Flags: true}).Collect()
|
||||
if err != nil {
|
||||
return fmt.Errorf("the folder %s's messages cannot be listed: %v", folder, err)
|
||||
}
|
||||
answered()
|
||||
for _, m := range all {
|
||||
server[uint32(m.UID)] = flagStrings(m.Flags)
|
||||
}
|
||||
}
|
||||
index, err := k.Store.Flags(ctx, a.Name, folder)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
say := func(e *Event) *Event {
|
||||
if seen {
|
||||
return e
|
||||
}
|
||||
return nil
|
||||
}
|
||||
// Removed: in the index, no longer on the server.
|
||||
for uid := range index {
|
||||
if _, still := server[uid]; still {
|
||||
continue
|
||||
}
|
||||
m, _, err := k.Store.Get(ctx, a.Name, folder, uid)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if m.UID == 0 {
|
||||
m = Message{Account: a.Name, Folder: folder, UID: uid, UIDValidity: sel.UIDValidity}
|
||||
}
|
||||
if err := k.Store.Remove(ctx, a.Name, folder, uid, say(removedEvent(m))); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
// Changed: on both, with other flags.
|
||||
var added []uint32
|
||||
for uid, flags := range server {
|
||||
before, indexed := index[uid]
|
||||
if !indexed {
|
||||
added = append(added, uid)
|
||||
continue
|
||||
}
|
||||
if sameStrings(before, flags) {
|
||||
continue
|
||||
}
|
||||
m, _, err := k.Store.Get(ctx, a.Name, folder, uid)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
m.Flags = flags
|
||||
if err := k.Store.SetFlags(ctx, a.Name, folder, uid, flags, say(changedEvent(m, before, flags))); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
// Added: on the server, not yet in the index. Their sizes first, then their bodies in batches bounded by bytes as
|
||||
// well as by count, so one batch never holds more than BatchBytes of mail in memory; a message larger than that
|
||||
// is fetched alone, and said. Fetched without marking them seen.
|
||||
sort.Slice(added, func(i, j int) bool { return added[i] < added[j] })
|
||||
sizes := map[uint32]int64{}
|
||||
for start := 0; start < len(added); start += SizeBatch {
|
||||
part := added[start:min(start+SizeBatch, len(added))]
|
||||
got, err := c.Fetch(uidSet(part), &imap.FetchOptions{UID: true, RFC822Size: true}).Collect()
|
||||
if err != nil {
|
||||
return fmt.Errorf("the folder %s's new messages cannot be measured: %v", folder, err)
|
||||
}
|
||||
answered()
|
||||
for _, m := range got {
|
||||
sizes[uint32(m.UID)] = m.RFC822Size
|
||||
}
|
||||
}
|
||||
cap := orBytes(k.BatchBytes, BatchBytes)
|
||||
for _, batch := range Batches(added, sizes, cap, FetchBatch) {
|
||||
if len(batch) == 1 && sizes[batch[0]] > cap {
|
||||
k.logf("[mesh-mail] the account %s's message %d in %s is %d bytes, more than a batch holds; it is fetched alone",
|
||||
a.Name, batch[0], folder, sizes[batch[0]])
|
||||
}
|
||||
msgs, err := c.Fetch(uidSet(batch), &imap.FetchOptions{
|
||||
UID: true, Flags: true, Envelope: true, RFC822Size: true, InternalDate: true,
|
||||
BodySection: []*imap.FetchItemBodySection{{Peek: true}},
|
||||
}).Collect()
|
||||
if err != nil {
|
||||
return fmt.Errorf("the folder %s's new messages cannot be fetched: %v", folder, err)
|
||||
}
|
||||
answered()
|
||||
for _, fm := range msgs {
|
||||
m, err := k.index(ctx, a, folder, sel.UIDValidity, fm)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if err := k.Store.Add(ctx, m, say(addedEvent(m))); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
}
|
||||
return k.Store.SetFolder(ctx, Folder{Account: a.Name, Name: folder, UIDValidity: sel.UIDValidity, Messages: len(server), Synced: k.now()})
|
||||
}
|
||||
|
||||
// index is a fetched message as the index keeps it: its metadata, its text, and its attachments kept in the store.
|
||||
func (k *Keeper) index(ctx context.Context, a Account, folder string, validity uint32, fm *imapclient.FetchMessageBuffer) (Message, error) {
|
||||
m := Message{Account: a.Name, Folder: folder, UIDValidity: validity, UID: uint32(fm.UID), Size: fm.RFC822Size,
|
||||
Flags: flagStrings(fm.Flags), Date: fm.InternalDate.UTC(), From: []Person{}, To: []Person{}, Attachments: []AttachmentRef{}}
|
||||
if env := fm.Envelope; env != nil {
|
||||
m.MessageID, m.Subject = env.MessageID, env.Subject
|
||||
if len(env.InReplyTo) > 0 {
|
||||
m.InReplyTo = env.InReplyTo[0]
|
||||
}
|
||||
if !env.Date.IsZero() {
|
||||
m.Date = env.Date.UTC()
|
||||
}
|
||||
m.From, m.To, m.Cc = people(env.From), people(env.To), people(env.Cc)
|
||||
}
|
||||
var raw []byte
|
||||
if len(fm.BodySection) > 0 {
|
||||
raw = fm.BodySection[0].Bytes
|
||||
}
|
||||
parsed, err := ParseMessage(raw)
|
||||
if err != nil {
|
||||
k.logf("[mesh-mail] the account %s's message %d in %s is read in part: %v", a.Name, m.UID, folder, err)
|
||||
}
|
||||
m.Body = parsed.Text
|
||||
for _, p := range parsed.Attachments {
|
||||
ref, err := Keep(ctx, k.Objects, p.Data, p.Type, p.Name)
|
||||
if err != nil {
|
||||
return m, err
|
||||
}
|
||||
m.Attachments = append(m.Attachments, ref)
|
||||
}
|
||||
return m, nil
|
||||
}
|
||||
|
||||
func people(as []imap.Address) []Person {
|
||||
out := []Person{}
|
||||
for _, a := range as {
|
||||
if addr := a.Addr(); addr != "" {
|
||||
out = append(out, Person{Name: a.Name, Address: addr})
|
||||
}
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
func flagStrings(fs []imap.Flag) []string {
|
||||
out := make([]string, 0, len(fs))
|
||||
for _, f := range fs {
|
||||
if strings.EqualFold(string(f), `\Recent`) {
|
||||
continue
|
||||
}
|
||||
out = append(out, string(f))
|
||||
}
|
||||
sort.Strings(out)
|
||||
return out
|
||||
}
|
||||
|
||||
func sameStrings(a, b []string) bool {
|
||||
if len(a) != len(b) {
|
||||
return false
|
||||
}
|
||||
for i := range a {
|
||||
if a[i] != b[i] {
|
||||
return false
|
||||
}
|
||||
}
|
||||
return true
|
||||
}
|
||||
|
||||
// ---- the events ---------------------------------------------------------------------------------------------------
|
||||
|
||||
// EventKey is where an account's change is said: mail.<account>.message.<added|changed|removed> (research 049
|
||||
// document 02 §5), the account by its name in the setting, never its address.
|
||||
func EventKey(account, change string) string { return "mail." + account + ".message." + change }
|
||||
|
||||
// Changes are the three things said of a message.
|
||||
var Changes = []string{"added", "changed", "removed"}
|
||||
|
||||
// metadata is what an event says of a message: never its body, never an attachment's bytes (ADR 0359 §2).
|
||||
func metadata(m Message) map[string]any {
|
||||
atts := make([]map[string]any, 0, len(m.Attachments))
|
||||
for _, r := range m.Attachments {
|
||||
atts = append(atts, map[string]any{"bucket": r.Bucket, "key": r.Key, "size": r.Size, "type": r.Type, "name": r.Name})
|
||||
}
|
||||
return map[string]any{
|
||||
"account": m.Account, "folder": m.Folder, "uid": m.UID, "uidvalidity": m.UIDValidity,
|
||||
"message-id": m.MessageID, "in-reply-to": m.InReplyTo, "date": m.Date.UTC().Format(time.RFC3339),
|
||||
"from": m.From, "to": m.To, "cc": m.Cc, "subject": m.Subject, "size": m.Size, "flags": m.Flags,
|
||||
"attachments": atts,
|
||||
}
|
||||
}
|
||||
|
||||
func addedEvent(m Message) *Event {
|
||||
return &Event{Key: EventKey(m.Account, "added"), Body: metadata(m)}
|
||||
}
|
||||
|
||||
func changedEvent(m Message, before, after []string) *Event {
|
||||
body := metadata(m)
|
||||
body["changes"] = map[string]any{"flags": map[string]any{"before": before, "after": after}}
|
||||
return &Event{Key: EventKey(m.Account, "changed"), Body: body}
|
||||
}
|
||||
|
||||
func removedEvent(m Message) *Event {
|
||||
return &Event{Key: EventKey(m.Account, "removed"), Body: map[string]any{
|
||||
"account": m.Account, "folder": m.Folder, "uid": m.UID, "uidvalidity": m.UIDValidity, "message-id": m.MessageID,
|
||||
}}
|
||||
}
|
||||
@@ -0,0 +1,292 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"crypto/sha256"
|
||||
"encoding/hex"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"regexp"
|
||||
"strings"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/emersion/go-imap/v2"
|
||||
)
|
||||
|
||||
// keeping starts a keeper on one account of the test server, its paces short, and stops it when the test ends.
|
||||
func keeping(t *testing.T, s *mailServer, st Store, objects Objects, secrets string) {
|
||||
t.Helper()
|
||||
k := &Keeper{Store: st, Objects: objects, Dial: s.dialer(), SecretsDir: secrets, Logf: t.Logf,
|
||||
IdleCycle: 300 * time.Millisecond, PollEvery: 400 * time.Millisecond, RetryFirst: 100 * time.Millisecond}
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
done := make(chan struct{})
|
||||
go func() { k.Watch(ctx, testAccount()); close(done) }()
|
||||
t.Cleanup(func() { cancel(); <-done })
|
||||
}
|
||||
|
||||
func stateOf(st Store, account string) AccountState {
|
||||
states, _ := st.States(context.Background())
|
||||
return states[account]
|
||||
}
|
||||
|
||||
// float64OrUint reads a uid from an event body whether it came from memory (uint32) or from postgres (float64).
|
||||
type float64OrUint struct{ v any }
|
||||
|
||||
func (f float64OrUint) value() uint32 {
|
||||
switch n := f.v.(type) {
|
||||
case uint32:
|
||||
return n
|
||||
case float64:
|
||||
return uint32(n)
|
||||
}
|
||||
return 0
|
||||
}
|
||||
|
||||
func uidOf(e Event) uint32 { return float64OrUint{e.Body["uid"]}.value() }
|
||||
|
||||
func eventsNamed(st Store, key string) []Event {
|
||||
evs, _ := st.Pending(context.Background(), 1000)
|
||||
var out []Event
|
||||
for _, e := range evs {
|
||||
if e.Key == key {
|
||||
out = append(out, e)
|
||||
}
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
func describe(st Store) string {
|
||||
evs, _ := st.Pending(context.Background(), 1000)
|
||||
var keys []string
|
||||
for _, e := range evs {
|
||||
keys = append(keys, e.Key)
|
||||
}
|
||||
s := stateOf(st, "mesh")
|
||||
return "state " + s.State + " (" + s.Why + "); events " + strings.Join(keys, ", ")
|
||||
}
|
||||
|
||||
// The whole watch, over a real IMAP conversation (ADR 0359 §2, §3, §5): the first look indexes what is there and says
|
||||
// nothing; then a new mail in the inbox is said within moments of its delivery, through IDLE, with its metadata and
|
||||
// its attachment by reference, never its body; its flags changed are said; its removal is said; a mail in another
|
||||
// folder is said at the next poll; and watching never marks a message seen.
|
||||
func TestTheKeeperSaysEachChangeOnTheBusAndOnlyReads(t *testing.T) {
|
||||
eachStore(t, func(t *testing.T, st Store) {
|
||||
s := newMailServer(t)
|
||||
op := s.operator(t)
|
||||
deliver(t, op, "INBOX", plainMail("before the watch", "an old letter"))
|
||||
deliver(t, op, "Archive", plainMail("archived long ago", "an older letter"))
|
||||
secrets := t.TempDir()
|
||||
giveSecret(t, secrets, testAccount(), testPassword)
|
||||
objects := newMemObjects()
|
||||
keeping(t, s, st, objects, secrets)
|
||||
|
||||
eventually(t, 10*time.Second, "the first look", func() (bool, string) {
|
||||
return stateOf(st, "mesh").State == AccountCurrent, describe(st)
|
||||
})
|
||||
ctx := context.Background()
|
||||
if hits, _ := st.Search(ctx, Query{Text: "letter", Accounts: []string{"mesh"}}); len(hits) != 2 {
|
||||
t.Fatalf("the first look indexed %d messages, not 2", len(hits))
|
||||
}
|
||||
if evs, _ := st.Pending(ctx, 100); len(evs) != 0 {
|
||||
t.Fatalf("the first look said %d event(s): the history of a mailbox is not news", len(evs))
|
||||
}
|
||||
|
||||
// A new mail with an attachment, delivered while the keeper idles on the inbox.
|
||||
deliver(t, op, "INBOX", attachedMail("the invoice"))
|
||||
var added Event
|
||||
eventually(t, 5*time.Second, "the new mail said", func() (bool, string) {
|
||||
evs := eventsNamed(st, "mail.mesh.message.added")
|
||||
if len(evs) == 1 {
|
||||
added = evs[0]
|
||||
}
|
||||
return len(evs) == 1, describe(st)
|
||||
})
|
||||
body := added.Body
|
||||
if body["subject"] != "the invoice" || body["folder"] != "INBOX" || body["account"] != "mesh" || body["message-id"] != "att@example.org" {
|
||||
t.Fatalf("the event says %v", body)
|
||||
}
|
||||
raw, _ := json.Marshal(body)
|
||||
for _, never := range []string{"Please find", "dGhlIGludm9pY2Ug", attachmentData} {
|
||||
if strings.Contains(string(raw), never) {
|
||||
t.Fatalf("the event carries the body or an attachment's bytes (%q): %s", never, raw)
|
||||
}
|
||||
}
|
||||
sum := sha256.Sum256([]byte(attachmentData))
|
||||
key := hex.EncodeToString(sum[:])
|
||||
if !strings.Contains(string(raw), `"key":"`+key+`"`) || !strings.Contains(string(raw), `"name":"invoice.pdf"`) {
|
||||
t.Fatalf("the event does not name the attachment by its SHA-256 %s: %s", key, raw)
|
||||
}
|
||||
if kept, _ := objects.Get(ctx, key); string(kept) != attachmentData {
|
||||
t.Fatalf("the attachment store keeps %q under %s", kept, key)
|
||||
}
|
||||
uid := uidOf(added)
|
||||
m, ok, _ := st.Get(ctx, "mesh", "INBOX", uid)
|
||||
if !ok || !strings.Contains(m.Body, "Please find the invoice attached.") {
|
||||
t.Fatalf("the index keeps the text for mail_read: %+v", m)
|
||||
}
|
||||
|
||||
// Flagged by the operator elsewhere: said as changed, with the flags before and after.
|
||||
if _, err := op.Select("INBOX", nil).Wait(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := op.Store(imap.UIDSetNum(imap.UID(uid)), &imap.StoreFlags{Op: imap.StoreFlagsAdd, Flags: []imap.Flag{imap.FlagFlagged}, Silent: true}, nil).Close(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
eventually(t, 5*time.Second, "the flag said", func() (bool, string) {
|
||||
evs := eventsNamed(st, "mail.mesh.message.changed")
|
||||
if len(evs) != 1 {
|
||||
return false, describe(st)
|
||||
}
|
||||
raw, _ := json.Marshal(evs[0].Body["changes"])
|
||||
return strings.Contains(string(raw), `"after":["\\Flagged"]`), string(raw)
|
||||
})
|
||||
|
||||
// Never marked seen by the watching: the message is still unseen on the server.
|
||||
fetched, err := op.Fetch(imap.UIDSetNum(imap.UID(uid)), &imap.FetchOptions{Flags: true, UID: true}).Collect()
|
||||
if err != nil || len(fetched) != 1 {
|
||||
t.Fatalf("%v %v", fetched, err)
|
||||
}
|
||||
for _, f := range fetched[0].Flags {
|
||||
if f == imap.FlagSeen {
|
||||
t.Fatal("watching marked the message seen")
|
||||
}
|
||||
}
|
||||
|
||||
// Deleted by the operator elsewhere: said as removed.
|
||||
if err := op.Store(imap.UIDSetNum(imap.UID(uid)), &imap.StoreFlags{Op: imap.StoreFlagsAdd, Flags: []imap.Flag{imap.FlagDeleted}, Silent: true}, nil).Close(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := op.Expunge().Close(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
eventually(t, 5*time.Second, "the removal said", func() (bool, string) {
|
||||
evs := eventsNamed(st, "mail.mesh.message.removed")
|
||||
return len(evs) == 1 && uidOf(evs[0]) == uid && evs[0].Body["message-id"] == "att@example.org", describe(st)
|
||||
})
|
||||
if _, ok, _ := st.Get(ctx, "mesh", "INBOX", uid); ok {
|
||||
t.Fatal("the removed message is still indexed")
|
||||
}
|
||||
|
||||
// A mail in another folder is said at the next poll.
|
||||
deliver(t, op, "Archive", plainMail("filed away", "a letter filed"))
|
||||
eventually(t, 5*time.Second, "the archived mail said", func() (bool, string) {
|
||||
for _, e := range eventsNamed(st, "mail.mesh.message.added") {
|
||||
if e.Body["folder"] == "Archive" && e.Body["subject"] == "filed away" {
|
||||
return true, ""
|
||||
}
|
||||
}
|
||||
return false, describe(st)
|
||||
})
|
||||
if s := stateOf(st, "mesh"); s.State != AccountCurrent || time.Since(s.Heard) > time.Minute {
|
||||
t.Fatalf("the account is %+v", s)
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
// No password given: the account waits, and nothing is dialled. A wrong one: the account fails, saying the server
|
||||
// refused the sign-in. Given later: it is picked up without a restart.
|
||||
func TestTheKeeperWaitsForThePasswordAndSaysARefusal(t *testing.T) {
|
||||
s := newMailServer(t)
|
||||
st := newMemStore()
|
||||
secrets := t.TempDir()
|
||||
keeping(t, s, st, newMemObjects(), secrets)
|
||||
eventually(t, 5*time.Second, "the wait", func() (bool, string) {
|
||||
got := stateOf(st, "mesh")
|
||||
return got.State == AccountWaits && strings.Contains(got.Why, "mail-credential-mesh"), describe(st)
|
||||
})
|
||||
giveSecret(t, secrets, testAccount(), "not-the-password")
|
||||
eventually(t, 5*time.Second, "the refusal", func() (bool, string) {
|
||||
got := stateOf(st, "mesh")
|
||||
return got.State == AccountFailing && strings.Contains(got.Why, "refused the sign-in"), describe(st)
|
||||
})
|
||||
giveSecret(t, secrets, testAccount(), testPassword)
|
||||
eventually(t, 10*time.Second, "the sign-in", func() (bool, string) {
|
||||
return stateOf(st, "mesh").State == AccountCurrent, describe(st)
|
||||
})
|
||||
}
|
||||
|
||||
// It only reads (ADR 0359 §5): no code of the module, but its tests, issues a command that flags, moves, copies,
|
||||
// deletes, appends, creates, renames or sends, and every folder is opened read-only and every body peeked.
|
||||
func TestNothingButTheTestsWritesToAMailbox(t *testing.T) {
|
||||
files, err := filepath.Glob("*.go")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
writes := regexp.MustCompile(`([A-Za-z_][A-Za-z0-9_]*)\.(Store|Move|Copy|Expunge|UIDExpunge|Append|Create|Delete|Rename|Subscribe|Unsubscribe|SetMetadata|SetQuota)\(`)
|
||||
for _, f := range files {
|
||||
if strings.HasSuffix(f, "_test.go") {
|
||||
continue
|
||||
}
|
||||
raw, err := os.ReadFile(f)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
for _, m := range writes.FindAllSubmatch(raw, -1) {
|
||||
if string(m[1]) != "os" { // a file of the module's own, never a mailbox
|
||||
t.Errorf("%s calls %s, which writes to a mailbox", f, m[0])
|
||||
}
|
||||
}
|
||||
if strings.Contains(string(raw), "net/smtp") {
|
||||
t.Errorf("%s speaks SMTP: this stage sends nothing", f)
|
||||
}
|
||||
}
|
||||
src, _ := os.ReadFile("imap.go")
|
||||
if !strings.Contains(string(src), "&imap.SelectOptions{ReadOnly: true}") || !strings.Contains(string(src), "{{Peek: true}}") {
|
||||
t.Fatal("folders are not opened read-only, or bodies not peeked")
|
||||
}
|
||||
}
|
||||
|
||||
// A fetch of bodies is bounded by bytes (issue 733): batches cut by size and by count, and a message larger than a
|
||||
// batch holds alone.
|
||||
func TestBatchesAreBoundedByBytesAndALargerMessageGoesAlone(t *testing.T) {
|
||||
sizes := map[uint32]int64{1: 40, 2: 40, 3: 500, 4: 10, 5: 10, 6: 10}
|
||||
got := Batches([]uint32{1, 2, 3, 4, 5, 6}, sizes, 100, 2)
|
||||
want := [][]uint32{{1, 2}, {3}, {4, 5}, {6}}
|
||||
if fmt.Sprint(got) != fmt.Sprint(want) {
|
||||
t.Fatalf("%v, not %v", got, want)
|
||||
}
|
||||
if got := Batches(nil, nil, 100, 25); len(got) != 0 {
|
||||
t.Fatalf("%v", got)
|
||||
}
|
||||
}
|
||||
|
||||
// Over a real conversation: with a cap of a few hundred bytes, a large mail is fetched alone and said so, and every
|
||||
// mail is still indexed and said.
|
||||
func TestTheKeeperFetchesUnderAByteCap(t *testing.T) {
|
||||
s := newMailServer(t)
|
||||
op := s.operator(t)
|
||||
st := newMemStore()
|
||||
secrets := t.TempDir()
|
||||
giveSecret(t, secrets, testAccount(), testPassword)
|
||||
var mu sync.Mutex
|
||||
var said []string
|
||||
k := &Keeper{Store: st, Objects: newMemObjects(), Dial: s.dialer(), SecretsDir: secrets,
|
||||
Logf: func(f string, a ...any) { mu.Lock(); said = append(said, fmt.Sprintf(f, a...)); mu.Unlock() },
|
||||
IdleCycle: 300 * time.Millisecond, PollEvery: 400 * time.Millisecond, RetryFirst: 100 * time.Millisecond, BatchBytes: 600}
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
done := make(chan struct{})
|
||||
go func() { k.Watch(ctx, testAccount()); close(done) }()
|
||||
t.Cleanup(func() { cancel(); <-done })
|
||||
eventually(t, 10*time.Second, "the first look", func() (bool, string) { return stateOf(st, "mesh").State == AccountCurrent, describe(st) })
|
||||
deliver(t, op, "INBOX", plainMail("small one", "short"))
|
||||
deliver(t, op, "INBOX", plainMail("large one", strings.Repeat("a long letter ", 200)))
|
||||
deliver(t, op, "INBOX", plainMail("small two", "short"))
|
||||
eventually(t, 5*time.Second, "three mails said", func() (bool, string) {
|
||||
return len(eventsNamed(st, "mail.mesh.message.added")) == 3, describe(st)
|
||||
})
|
||||
mu.Lock()
|
||||
defer mu.Unlock()
|
||||
alone := 0
|
||||
for _, line := range said {
|
||||
if strings.Contains(line, "fetched alone") {
|
||||
alone++
|
||||
}
|
||||
}
|
||||
if alone != 1 {
|
||||
t.Fatalf("said %d message(s) fetched alone: %v", alone, said)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,252 @@
|
||||
package main
|
||||
|
||||
// What the tests stand on: an IMAP server in this process (go-imap's own in-memory server, speaking the protocol over
|
||||
// a real socket), the index in memory always and on a real postgres when one can be had (MESH_TEST_POSTGRES, or a
|
||||
// private server started here from the machine's initdb on a socket of its own, listening on no port), and the
|
||||
// attachment store in memory.
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"net"
|
||||
"net/url"
|
||||
"os"
|
||||
"os/exec"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/emersion/go-imap/v2"
|
||||
"github.com/emersion/go-imap/v2/imapclient"
|
||||
"github.com/emersion/go-imap/v2/imapserver"
|
||||
"github.com/emersion/go-imap/v2/imapserver/imapmemserver"
|
||||
)
|
||||
|
||||
func TestMain(m *testing.M) {
|
||||
code := m.Run()
|
||||
stopPostgres()
|
||||
os.Exit(code)
|
||||
}
|
||||
|
||||
// ---- the IMAP server --------------------------------------------------------------------------------------------
|
||||
|
||||
const (
|
||||
testUser = "operator@example.org"
|
||||
testPassword = "the-password"
|
||||
)
|
||||
|
||||
type mailServer struct {
|
||||
addr string
|
||||
srv *imapserver.Server
|
||||
}
|
||||
|
||||
// newMailServer is a server holding one user with an inbox and an Archive folder.
|
||||
func newMailServer(t *testing.T) *mailServer {
|
||||
t.Helper()
|
||||
mem := imapmemserver.New()
|
||||
user := imapmemserver.NewUser(testUser, testPassword)
|
||||
for _, f := range []string{"INBOX", "Archive"} {
|
||||
if err := user.Create(f, nil); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
mem.AddUser(user)
|
||||
srv := imapserver.New(&imapserver.Options{
|
||||
NewSession: func(*imapserver.Conn) (imapserver.Session, *imapserver.GreetingData, error) {
|
||||
return mem.NewSession(), nil, nil
|
||||
},
|
||||
InsecureAuth: true,
|
||||
Caps: imap.CapSet{imap.CapIMAP4rev1: {}, imap.CapIMAP4rev2: {}},
|
||||
})
|
||||
ln, err := net.Listen("tcp", "127.0.0.1:0")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
go func() { _ = srv.Serve(ln) }()
|
||||
t.Cleanup(func() { _ = srv.Close() })
|
||||
return &mailServer{addr: ln.Addr().String(), srv: srv}
|
||||
}
|
||||
|
||||
// dialer reaches the test server without TLS, which only a test does: the mesh's dialer signs in over TLS alone.
|
||||
func (s *mailServer) dialer() Dialer {
|
||||
return func(_ Account, h *imapclient.UnilateralDataHandler) (reader, error) {
|
||||
c, err := imapclient.DialInsecure(s.addr, &imapclient.Options{UnilateralDataHandler: h})
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return c, nil
|
||||
}
|
||||
}
|
||||
|
||||
// operator is a second client, as the operator's mail program or another sender is: it appends, flags and deletes.
|
||||
func (s *mailServer) operator(t *testing.T) *imapclient.Client {
|
||||
t.Helper()
|
||||
c, err := imapclient.DialInsecure(s.addr, nil)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
t.Cleanup(func() { _ = c.Close() })
|
||||
if err := c.Login(testUser, testPassword).Wait(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return c
|
||||
}
|
||||
|
||||
func deliver(t *testing.T, c *imapclient.Client, folder, raw string) {
|
||||
t.Helper()
|
||||
raw = strings.ReplaceAll(raw, "\n", "\r\n")
|
||||
cmd := c.Append(folder, int64(len(raw)), nil)
|
||||
if _, err := cmd.Write([]byte(raw)); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := cmd.Close(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := cmd.Wait(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
|
||||
func plainMail(subject, body string) string {
|
||||
return "From: Alice <alice@example.org>\nTo: Operator <operator@example.org>\nSubject: " + subject +
|
||||
"\nMessage-Id: <" + strings.ReplaceAll(subject, " ", "-") + "@example.org>\nDate: Wed, 14 Oct 2026 10:00:00 +0000\n" +
|
||||
"MIME-Version: 1.0\nContent-Type: text/plain; charset=utf-8\n\n" + body + "\n"
|
||||
}
|
||||
|
||||
// attachmentData is an attachment's content, and attachedMail a message carrying it.
|
||||
const attachmentData = "the invoice, in full"
|
||||
|
||||
func attachedMail(subject string) string {
|
||||
return "From: Bob <bob@example.org>\nTo: operator@example.org\nSubject: " + subject + "\nMessage-Id: <att@example.org>\n" +
|
||||
"Date: Wed, 14 Oct 2026 11:00:00 +0000\nMIME-Version: 1.0\nContent-Type: multipart/mixed; boundary=XYZ\n\n" +
|
||||
"--XYZ\nContent-Type: text/plain; charset=utf-8\n\nPlease find the invoice attached.\n" +
|
||||
"--XYZ\nContent-Type: application/pdf; name=invoice.pdf\nContent-Disposition: attachment; filename=invoice.pdf\n" +
|
||||
"Content-Transfer-Encoding: base64\n\ndGhlIGludm9pY2UsIGluIGZ1bGw=\n--XYZ--\n"
|
||||
}
|
||||
|
||||
// ---- the account ------------------------------------------------------------------------------------------------
|
||||
|
||||
func testAccount() Account {
|
||||
return Account{Name: "mesh", Kind: KindIMAP, Address: testUser, Incoming: Server{Host: "localhost", Port: 993, Security: "tls"}}
|
||||
}
|
||||
|
||||
func giveSecret(t *testing.T, dir string, a Account, pw string) {
|
||||
t.Helper()
|
||||
if err := os.WriteFile(SecretPath(dir, a), []byte(pw+"\n"), 0o600); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
|
||||
// eventually waits for a condition, failing with what it last saw.
|
||||
func eventually(t *testing.T, within time.Duration, what string, ok func() (bool, string)) {
|
||||
t.Helper()
|
||||
deadline := time.Now().Add(within)
|
||||
last := ""
|
||||
for time.Now().Before(deadline) {
|
||||
var done bool
|
||||
if done, last = ok(); done {
|
||||
return
|
||||
}
|
||||
time.Sleep(20 * time.Millisecond)
|
||||
}
|
||||
t.Fatalf("%s did not happen within %s; last seen: %s", what, within, last)
|
||||
}
|
||||
|
||||
// ---- postgres ---------------------------------------------------------------------------------------------------
|
||||
|
||||
var (
|
||||
pgOnce sync.Once
|
||||
pgBase string
|
||||
pgDir string
|
||||
pgWhy string
|
||||
pgCount int
|
||||
pgMu sync.Mutex
|
||||
)
|
||||
|
||||
func startPostgres() {
|
||||
if u := os.Getenv("MESH_TEST_POSTGRES"); u != "" {
|
||||
pgBase = u
|
||||
return
|
||||
}
|
||||
initdb, err := exec.LookPath("initdb")
|
||||
if err != nil {
|
||||
pgWhy = "no postgres: MESH_TEST_POSTGRES is unset and the machine has no initdb"
|
||||
return
|
||||
}
|
||||
pgctl := filepath.Join(filepath.Dir(initdb), "pg_ctl")
|
||||
dir, err := os.MkdirTemp("", "mmpg")
|
||||
if err != nil {
|
||||
pgWhy = err.Error()
|
||||
return
|
||||
}
|
||||
pgDir = dir
|
||||
if out, err := exec.Command(initdb, "-D", dir+"/data", "-U", "postgres", "-A", "trust", "-E", "UTF8", "--no-locale").CombinedOutput(); err != nil {
|
||||
pgWhy = fmt.Sprintf("initdb: %v: %s", err, out)
|
||||
return
|
||||
}
|
||||
if out, err := exec.Command(pgctl, "-D", dir+"/data", "-l", dir+"/log", "-w", "-o",
|
||||
"-k "+dir+" -c listen_addresses='' -F", "start").CombinedOutput(); err != nil {
|
||||
pgWhy = fmt.Sprintf("pg_ctl start: %v: %s", err, out)
|
||||
return
|
||||
}
|
||||
pgBase = "postgres://postgres@/postgres?host=" + dir
|
||||
}
|
||||
|
||||
func stopPostgres() {
|
||||
if pgDir == "" {
|
||||
return
|
||||
}
|
||||
if initdb, err := exec.LookPath("initdb"); err == nil {
|
||||
_ = exec.Command(filepath.Join(filepath.Dir(initdb), "pg_ctl"), "-D", pgDir+"/data", "-m", "immediate", "stop").Run()
|
||||
}
|
||||
_ = os.RemoveAll(pgDir)
|
||||
}
|
||||
|
||||
// freshDatabase is a database of its own for one test, or a skip saying why there is none.
|
||||
func freshDatabase(t *testing.T) string {
|
||||
t.Helper()
|
||||
pgOnce.Do(startPostgres)
|
||||
if pgBase == "" {
|
||||
t.Skipf("NOT TESTED ON POSTGRES: %s", pgWhy)
|
||||
}
|
||||
pgMu.Lock()
|
||||
pgCount++
|
||||
name := fmt.Sprintf("mail_%d_%d", os.Getpid(), pgCount)
|
||||
pgMu.Unlock()
|
||||
ctx := context.Background()
|
||||
admin, err := OpenPgStore(ctx, pgBase)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer admin.Close()
|
||||
if _, err := admin.pool.Exec(ctx, "create database "+name); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
u, err := url.Parse(pgBase)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
u.Path = "/" + name
|
||||
return u.String()
|
||||
}
|
||||
|
||||
func pgStore(t *testing.T) *PgStore {
|
||||
t.Helper()
|
||||
s, err := OpenPgStore(context.Background(), freshDatabase(t))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
t.Cleanup(s.Close)
|
||||
if _, err := s.Prepare(context.Background()); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return s
|
||||
}
|
||||
|
||||
// eachStore runs a test over the index in memory and on postgres.
|
||||
func eachStore(t *testing.T, run func(t *testing.T, s Store)) {
|
||||
t.Run("memory", func(t *testing.T) { run(t, newMemStore()) })
|
||||
t.Run("postgres", func(t *testing.T) { run(t, pgStore(t)) })
|
||||
}
|
||||
@@ -0,0 +1,250 @@
|
||||
// mesh-mail: the operator's mail accounts, watched, and each mail said on the bus (novox/hq ADR 0359). One instance,
|
||||
// on the control-node. The first stage holds the mesh's own server, read-only: no flag, move, delete, draft or send.
|
||||
//
|
||||
// mesh-mail the bundle the node's runtime launches: the tools mail_accounts, mail_search, mail_read and
|
||||
// mail_health, and the events the keeper wrote, said on the bus as the module
|
||||
// mesh-mail prepare the preparation step: the index's schema brought to this version's
|
||||
// mesh-mail keep acts: watches every account (IDLE on the inbox, a look at the other folders every five
|
||||
// minutes), keeps the index and the attachment store, and writes each change's event beside it
|
||||
// mesh-mail watch judges: reads every minute how each account is, and leaves its verdict for mail_health
|
||||
//
|
||||
// The two long-running processes are the act and the judge, as mesh-knowledge's are: keep is judged by its unit, watch
|
||||
// by the tool mail_health (ADR 0359 §6, ADR 0240 §3 and §7). Only the bundle reaches the bus, through the runtime that
|
||||
// launched it, so the keeper hands its events over in the index, written in the same step as the change they say,
|
||||
// and the bundle says them and removes each once the bus took it. stdout is the bundle's MCP channel; what this module
|
||||
// says, it says on stderr.
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"os"
|
||||
"os/signal"
|
||||
"sync"
|
||||
"syscall"
|
||||
"time"
|
||||
|
||||
stdio "git.novox.be/novox/mesh-sdk/go"
|
||||
)
|
||||
|
||||
// Module is this module's name.
|
||||
const Module = "mesh-mail"
|
||||
|
||||
// Env is where the mesh puts what the module reads, as the manifest's environment names it.
|
||||
type Env struct {
|
||||
Accounts string // the trusted setting accounts
|
||||
Grants string // the trusted setting grants
|
||||
Secrets string // the directory the own secrets mail-credential-<account> are placed in
|
||||
Store string // the attachment store's facts
|
||||
Status string // the watch's verdict
|
||||
}
|
||||
|
||||
func envFromOS() Env {
|
||||
return Env{
|
||||
Accounts: os.Getenv("MESH_MAIL_ACCOUNTS"),
|
||||
Grants: os.Getenv("MESH_MAIL_GRANTS"),
|
||||
Secrets: os.Getenv("MESH_MAIL_SECRETS"),
|
||||
Store: os.Getenv("MESH_MAIL_STORE"),
|
||||
Status: os.Getenv("MESH_MAIL_STATUS"),
|
||||
}
|
||||
}
|
||||
|
||||
func logf(format string, a ...any) { fmt.Fprintf(os.Stderr, format+"\n", a...) }
|
||||
|
||||
func main() {
|
||||
if len(os.Args) > 1 {
|
||||
switch os.Args[1] {
|
||||
case "prepare":
|
||||
os.Exit(prepare())
|
||||
case "keep":
|
||||
os.Exit(keep(envFromOS()))
|
||||
case "watch":
|
||||
os.Exit(watch(envFromOS()))
|
||||
}
|
||||
}
|
||||
serve(envFromOS())
|
||||
}
|
||||
|
||||
// ---- prepare ------------------------------------------------------------------------------------------------------
|
||||
|
||||
func prepare() int {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute)
|
||||
defer cancel()
|
||||
s, err := PgStoreFromEnv(ctx)
|
||||
if err != nil {
|
||||
logf("[mesh-mail] preparing the index: %v", err)
|
||||
return 1
|
||||
}
|
||||
defer s.Close()
|
||||
said, err := s.Prepare(ctx)
|
||||
for _, line := range said {
|
||||
logf("[mesh-mail] %s", line)
|
||||
}
|
||||
if err != nil {
|
||||
logf("[mesh-mail] preparing the index failed: %v", err)
|
||||
return 1
|
||||
}
|
||||
return 0
|
||||
}
|
||||
|
||||
// ---- keep ---------------------------------------------------------------------------------------------------------
|
||||
|
||||
// keep watches every account until it is stopped. A setting it cannot read, an index that is behind or a store it
|
||||
// cannot reach ends it with a fault, which its unit says; the mesh starts it again.
|
||||
func keep(env Env) int {
|
||||
ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
|
||||
defer stop()
|
||||
accounts, err := ReadAccounts(env.Accounts)
|
||||
if err != nil {
|
||||
logf("[mesh-mail] %v", err)
|
||||
return 1
|
||||
}
|
||||
st, err := PgStoreFromEnv(ctx)
|
||||
if err != nil {
|
||||
logf("[mesh-mail] the index cannot be opened: %v", err)
|
||||
return 1
|
||||
}
|
||||
defer st.Close()
|
||||
if err := st.Ready(ctx); err != nil {
|
||||
logf("[mesh-mail] %v", err)
|
||||
return 1
|
||||
}
|
||||
facts, err := ReadStoreFacts(env.Store)
|
||||
if err != nil {
|
||||
logf("[mesh-mail] %v", err)
|
||||
return 1
|
||||
}
|
||||
k := &Keeper{Store: st, Objects: NewS3(facts), Dial: DialServer, SecretsDir: env.Secrets, Logf: logf}
|
||||
logf("[mesh-mail] keeping %d account(s)", len(accounts))
|
||||
k.Run(ctx, accounts)
|
||||
return 0
|
||||
}
|
||||
|
||||
// ---- watch --------------------------------------------------------------------------------------------------------
|
||||
|
||||
func watch(env Env) int {
|
||||
if env.Status == "" {
|
||||
logf("[mesh-mail] watch was started without MESH_MAIL_STATUS")
|
||||
return 2
|
||||
}
|
||||
open := func(ctx context.Context) (Store, func(), error) {
|
||||
s, err := PgStoreFromEnv(ctx)
|
||||
if err != nil {
|
||||
return nil, nil, err
|
||||
}
|
||||
return s, s.Close, nil
|
||||
}
|
||||
for {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
|
||||
now := time.Now().UTC()
|
||||
answer := watchOnce(ctx, env, open, now)
|
||||
cancel()
|
||||
if err := writeStatus(env.Status, Status{Answer: answer, At: now}); err != nil {
|
||||
logf("[mesh-mail] the verdict could not be written: %v", err)
|
||||
}
|
||||
time.Sleep(WatchEvery)
|
||||
}
|
||||
}
|
||||
|
||||
// ---- the bundle ---------------------------------------------------------------------------------------------------
|
||||
|
||||
func serve(env Env) {
|
||||
var mu sync.Mutex
|
||||
var st *PgStore
|
||||
b := &Bundle{
|
||||
Env: env,
|
||||
Index: func(ctx context.Context) (Store, error) {
|
||||
mu.Lock()
|
||||
defer mu.Unlock()
|
||||
if st == nil {
|
||||
s, err := PgStoreFromEnv(ctx)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
st = s
|
||||
}
|
||||
if err := st.Ready(ctx); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return st, nil
|
||||
},
|
||||
Objects: func() (Objects, error) {
|
||||
facts, err := ReadStoreFacts(env.Store)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return NewS3(facts), nil
|
||||
},
|
||||
}
|
||||
p := &Publisher{Index: b.Index, Emit: func(key string, body any) error { return stdio.Emit(key, body) }, Logf: logf}
|
||||
go p.Run(context.Background())
|
||||
if err := stdio.Serve("", b.Tools()); err != nil {
|
||||
logf("%v", err)
|
||||
os.Exit(1)
|
||||
}
|
||||
}
|
||||
|
||||
// Publisher says the events the keeper wrote, oldest first, and removes each once the bus took it. One the bus did not
|
||||
// take stays, and is said again on the next round: delivery is at least once, and a consumer reads the uid and the
|
||||
// folder to know a repeat.
|
||||
type Publisher struct {
|
||||
Index func(ctx context.Context) (Store, error)
|
||||
Emit func(key string, body any) error
|
||||
Logf func(format string, a ...any)
|
||||
Every time.Duration
|
||||
}
|
||||
|
||||
// PublishBatch is how many events one round says at most.
|
||||
const PublishBatch = 100
|
||||
|
||||
// Once says what is pending, and answers how many it said.
|
||||
func (p *Publisher) Once(ctx context.Context) (int, error) {
|
||||
st, err := p.Index(ctx)
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
pending, err := st.Pending(ctx, PublishBatch)
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
var said []int64
|
||||
var failed error
|
||||
for _, e := range pending {
|
||||
if err := p.Emit(e.Key, e.Body); err != nil {
|
||||
failed = fmt.Errorf("the bus did not take %s: %v", e.Key, err)
|
||||
break
|
||||
}
|
||||
said = append(said, e.ID)
|
||||
}
|
||||
if len(said) > 0 {
|
||||
if err := st.Said(ctx, said); err != nil {
|
||||
return 0, err
|
||||
}
|
||||
}
|
||||
return len(said), failed
|
||||
}
|
||||
|
||||
// Run says pending events until ctx ends: every few seconds, at once again while a round was full.
|
||||
func (p *Publisher) Run(ctx context.Context) {
|
||||
every := p.Every
|
||||
if every <= 0 {
|
||||
every = 3 * time.Second
|
||||
}
|
||||
// The runtime takes the bundle's first calls only once Serve runs.
|
||||
time.Sleep(every)
|
||||
for ctx.Err() == nil {
|
||||
c, cancel := context.WithTimeout(ctx, time.Minute)
|
||||
n, err := p.Once(c)
|
||||
cancel()
|
||||
if err != nil && p.Logf != nil {
|
||||
p.Logf("[mesh-mail] events wait to be said: %v", err)
|
||||
}
|
||||
if n == PublishBatch {
|
||||
continue
|
||||
}
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
case <-time.After(every):
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,151 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"reflect"
|
||||
"strings"
|
||||
"testing"
|
||||
)
|
||||
|
||||
// The manifest says what the code does (novox/hq ADR 0359): the database and the bucket it requires (§3), the own
|
||||
// secret family issued outside the mesh (§5, ADR 0283 §6), the two trusted settings (§4), the tools (§5), the events,
|
||||
// and its two long-running processes with their health (§6) — and it is assigned to no machine by itself.
|
||||
|
||||
type manifest struct {
|
||||
Module string `json:"module"`
|
||||
Slug string `json:"slug"`
|
||||
Requires []string `json:"requires"`
|
||||
Contributes map[string]map[string]any `json:"contributes"`
|
||||
OwnSecrets map[string]map[string]string `json:"own-secrets"`
|
||||
Tools []string `json:"tools"`
|
||||
Emits []string `json:"emits"`
|
||||
Streams []struct {
|
||||
Name string `json:"name"`
|
||||
Events []string `json:"events"`
|
||||
MaxAgeDays int `json:"max-age-days"`
|
||||
Grant struct {
|
||||
Config string `json:"config"`
|
||||
Events string `json:"events"`
|
||||
} `json:"grant"`
|
||||
} `json:"streams"`
|
||||
Consumes []string `json:"consumes"`
|
||||
Invokes []string `json:"invokes"`
|
||||
Claims []json.RawMessage `json:"claims"`
|
||||
Resources []map[string]any `json:"resources"`
|
||||
Build struct {
|
||||
Artifacts []struct {
|
||||
From string `json:"from"`
|
||||
Binary string `json:"binary"`
|
||||
Loads []string `json:"loads"`
|
||||
Env map[string]string `json:"env"`
|
||||
} `json:"artifacts"`
|
||||
} `json:"build"`
|
||||
}
|
||||
|
||||
func readManifest(t *testing.T) (manifest, map[string]map[string]any) {
|
||||
t.Helper()
|
||||
raw, err := os.ReadFile(filepath.Join("..", "..", "module.json"))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
var m manifest
|
||||
if err := json.Unmarshal(raw, &m); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
byID := map[string]map[string]any{}
|
||||
for _, r := range m.Resources {
|
||||
byID[r["id"].(string)] = r
|
||||
}
|
||||
return m, byID
|
||||
}
|
||||
|
||||
func TestTheManifestSaysWhatTheCodeDoes(t *testing.T) {
|
||||
m, byID := readManifest(t)
|
||||
if m.Module != Module || !reflect.DeepEqual(m.Requires, []string{"postgres-database", "s3-bucket"}) || m.Contributes["postgres-database"]["name"] != "mail" {
|
||||
t.Fatalf("module %s, requires %v, contributes %v", m.Module, m.Requires, m.Contributes)
|
||||
}
|
||||
// mesh_<node>_<slug> fits the S3 access key's 20 characters on every node.
|
||||
if len("mesh_shanks_"+m.Slug) > 20 {
|
||||
t.Fatalf("the slug %q overflows an S3 access key", m.Slug)
|
||||
}
|
||||
fam := m.OwnSecrets[SecretFamily+"*"]
|
||||
if fam["issued-by"] != "outside" || fam["path"] != "${dir:state}/"+SecretFamily+"*.secret" || len(m.OwnSecrets) != 1 {
|
||||
t.Fatalf("own secrets %v", m.OwnSecrets)
|
||||
}
|
||||
var names []string
|
||||
for _, tool := range (&Bundle{}).Tools() {
|
||||
names = append(names, tool.Name)
|
||||
}
|
||||
if !reflect.DeepEqual(m.Tools, names) || !reflect.DeepEqual(names, []string{"mail_accounts", "mail_search", "mail_read", "mail_health"}) {
|
||||
t.Fatalf("tools %v, the code serves %v", m.Tools, names)
|
||||
}
|
||||
// Its events go to its own stream MAIL, kept a week, each account's part heard only by the consumers the trusted
|
||||
// config value grants names for it (ADR 0359 §2, §4; ADR 0363): none on the shared EVENTS stream.
|
||||
var events []string
|
||||
for _, c := range Changes {
|
||||
events = append(events, EventKey("*", c))
|
||||
}
|
||||
if len(m.Emits) != 0 || len(m.Streams) != 1 || m.Streams[0].Name != "mail" || !reflect.DeepEqual(m.Streams[0].Events, events) ||
|
||||
m.Streams[0].MaxAgeDays != 7 || m.Streams[0].Grant.Config != "grants" || m.Streams[0].Grant.Events != "mail.{key}.**" {
|
||||
t.Fatalf("emits %v, streams %+v; the code says %v", m.Emits, m.Streams, events)
|
||||
}
|
||||
// It reads its own mail and calls nothing, consumes nothing and claims no seat.
|
||||
if len(m.Consumes)+len(m.Invokes)+len(m.Claims) != 0 {
|
||||
t.Fatalf("consumes %v, invokes %v, claims %v", m.Consumes, m.Invokes, m.Claims)
|
||||
}
|
||||
// The two settings are trusted, and their defaults are an empty account list and no grant.
|
||||
for id, want := range map[string]string{"accounts": `{"accounts":{}}`, "grants": `{"grants":{}}`} {
|
||||
r := byID[id]
|
||||
if r == nil || r["trusted"] != true || r["merge"] != "json" || !reflect.DeepEqual(r["takes-settings"], []any{id}) {
|
||||
t.Fatalf("%s: %v", id, r)
|
||||
}
|
||||
if strings.Join(strings.Fields(r["content"].(string)), "") != want {
|
||||
t.Fatalf("%s's default is %q", id, r["content"])
|
||||
}
|
||||
if _, err := ParseAccounts([]byte(byID["accounts"]["content"].(string))); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
a := m.Build.Artifacts
|
||||
if len(a) != 1 || a[0].From != "cmd/mesh-mail" || a[0].Binary != Module || !reflect.DeepEqual(a[0].Loads, []string{Module}) {
|
||||
t.Fatalf("build %+v", m.Build)
|
||||
}
|
||||
for k, v := range map[string]string{"MESH_MAIL_ACCOUNTS": "accounts", "MESH_MAIL_GRANTS": "grants", "MESH_MAIL_STORE": "store", "DATABASE_URL_FILE": "database-url"} {
|
||||
if a[0].Env[k] != byID[v]["path"] {
|
||||
t.Errorf("the bundle reads %s at %s, and the manifest writes %s at %v", k, a[0].Env[k], v, byID[v]["path"])
|
||||
}
|
||||
}
|
||||
if a[0].Env["MESH_MAIL_SECRETS"] != "${dir:state}" {
|
||||
t.Errorf("the secrets directory is %s", a[0].Env["MESH_MAIL_SECRETS"])
|
||||
}
|
||||
}
|
||||
|
||||
// Its processes: the preparation once, keep judged by its unit, watch by mail_health (ADR 0359 §6), the watch writing
|
||||
// the verdict the bundle's tool reads, as the account the bundle runs as.
|
||||
func TestItsProcessesAreThePreparationTheActAndTheJudge(t *testing.T) {
|
||||
m, byID := readManifest(t)
|
||||
prep, keep, watch := byID["prepare"], byID["keep"], byID["watch"]
|
||||
if prep == nil || prep["run-once"] != true || prep["run"].([]any)[1] != "prepare" {
|
||||
t.Fatalf("prepare %v", prep)
|
||||
}
|
||||
if keep == nil || keep["run"].([]any)[1] != "keep" || keep["health"].(map[string]any)["kind"] != "unit" {
|
||||
t.Fatalf("keep %v", keep)
|
||||
}
|
||||
h, _ := watch["health"].(map[string]any)
|
||||
if watch == nil || watch["run"].([]any)[1] != "watch" || h["kind"] != "tool" || h["tool"] != HealthTool || h["needs"] != "postgres-database" {
|
||||
t.Fatalf("watch %v", watch)
|
||||
}
|
||||
if watch["user"] != "${machine:account}" || watch["env"].(map[string]any)["MESH_MAIL_STATUS"] != m.Build.Artifacts[0].Env["MESH_MAIL_STATUS"] {
|
||||
t.Fatalf("watch writes another status than the tool reads, or as another account: %v", watch)
|
||||
}
|
||||
// keep runs as the machine's root, which the own secrets' files belong to; it starts again when its settings or
|
||||
// its store change.
|
||||
if _, as := keep["user"]; as {
|
||||
t.Fatalf("keep runs as %v", keep["user"])
|
||||
}
|
||||
if !reflect.DeepEqual(keep["restart-on"], []any{"database-url", "store", "accounts"}) {
|
||||
t.Fatalf("keep restarts on %v", keep["restart-on"])
|
||||
}
|
||||
}
|
||||
+46
@@ -0,0 +1,46 @@
|
||||
-- mesh-mail's index (novox/hq ADR 0359 §3): one row per message and folder, what is known of each folder, each
|
||||
-- account's state as its keeper said it, and the events written with the change they say, until they are said.
|
||||
create table folder (
|
||||
account text not null,
|
||||
name text not null,
|
||||
uidvalidity bigint not null,
|
||||
messages integer not null default 0,
|
||||
synced timestamptz not null default now(),
|
||||
primary key (account, name));
|
||||
|
||||
create table message (
|
||||
account text not null,
|
||||
folder text not null,
|
||||
uid bigint not null,
|
||||
uidvalidity bigint not null,
|
||||
message_id text not null default '',
|
||||
in_reply_to text not null default '',
|
||||
date timestamptz not null,
|
||||
sender jsonb not null default '[]',
|
||||
recipients jsonb not null default '[]',
|
||||
copied jsonb not null default '[]',
|
||||
people text not null default '',
|
||||
subject text not null default '',
|
||||
size bigint not null default 0,
|
||||
flags text[] not null default '{}',
|
||||
attachments jsonb not null default '[]',
|
||||
body text not null default '',
|
||||
search tsvector generated always as (
|
||||
to_tsvector('simple'::regconfig, subject || ' ' || people || ' ' || body)) stored,
|
||||
primary key (account, folder, uid));
|
||||
create index message_search on message using gin (search);
|
||||
create index message_date on message (account, date desc);
|
||||
|
||||
create table account_state (
|
||||
account text primary key,
|
||||
state text not null,
|
||||
why text not null default '',
|
||||
heard timestamptz,
|
||||
synced timestamptz,
|
||||
said timestamptz not null default now());
|
||||
|
||||
create table outbox (
|
||||
id bigserial primary key,
|
||||
key text not null,
|
||||
body jsonb not null,
|
||||
made timestamptz not null default now());
|
||||
@@ -0,0 +1,125 @@
|
||||
package main
|
||||
|
||||
// A message's text and its attachments, read from its MIME source. The text is kept in the index for search and served
|
||||
// through mail_read; each attachment goes to the attachment store once, by its SHA-256 (novox/hq ADR 0359 §3).
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"errors"
|
||||
"html"
|
||||
"io"
|
||||
"mime"
|
||||
"regexp"
|
||||
"strings"
|
||||
|
||||
_ "github.com/emersion/go-message/charset" // the charsets mail is written in, beyond UTF-8 and ASCII
|
||||
"github.com/emersion/go-message/mail"
|
||||
)
|
||||
|
||||
// Part is one attachment as read from a message.
|
||||
type Part struct {
|
||||
Name string
|
||||
Type string
|
||||
Data []byte
|
||||
}
|
||||
|
||||
// MostBodyText bounds the text kept per message: a longer body is cut, and says so.
|
||||
const MostBodyText = 256 << 10
|
||||
|
||||
// Parsed is a message's text and its attachments.
|
||||
type Parsed struct {
|
||||
Text string
|
||||
Attachments []Part
|
||||
}
|
||||
|
||||
// ParseMessage reads a message's text (its plain part; else its HTML part as text) and its attachments. A message the
|
||||
// reader cannot follow keeps what was read before the fault, and the error says what.
|
||||
func ParseMessage(raw []byte) (Parsed, error) {
|
||||
var p Parsed
|
||||
r, err := mail.CreateReader(bytes.NewReader(raw))
|
||||
if r == nil {
|
||||
return p, err
|
||||
}
|
||||
var plain, htmlText string
|
||||
for {
|
||||
part, err := r.NextPart()
|
||||
if errors.Is(err, io.EOF) {
|
||||
break
|
||||
}
|
||||
if err != nil {
|
||||
return finish(p, plain, htmlText), err
|
||||
}
|
||||
switch h := part.Header.(type) {
|
||||
case *mail.InlineHeader:
|
||||
ct, params, _ := h.ContentType()
|
||||
_, disp, _ := h.ContentDisposition()
|
||||
body, readErr := io.ReadAll(io.LimitReader(part.Body, 64<<20))
|
||||
if readErr != nil {
|
||||
return finish(p, plain, htmlText), readErr
|
||||
}
|
||||
name := params["name"]
|
||||
if name == "" {
|
||||
name = disp["filename"]
|
||||
}
|
||||
switch {
|
||||
case ct == "text/plain" && plain == "" && name == "":
|
||||
plain = string(body)
|
||||
case ct == "text/html" && htmlText == "" && name == "":
|
||||
htmlText = HTMLText(string(body))
|
||||
case !strings.HasPrefix(ct, "text/") || name != "":
|
||||
// An inline image or a named inline file is an attachment too.
|
||||
p.Attachments = append(p.Attachments, Part{Name: name, Type: orOctets(ct), Data: body})
|
||||
}
|
||||
case *mail.AttachmentHeader:
|
||||
ct, _, _ := h.ContentType()
|
||||
name, _ := h.Filename()
|
||||
body, readErr := io.ReadAll(io.LimitReader(part.Body, 64<<20))
|
||||
if readErr != nil {
|
||||
return finish(p, plain, htmlText), readErr
|
||||
}
|
||||
p.Attachments = append(p.Attachments, Part{Name: name, Type: orOctets(ct), Data: body})
|
||||
}
|
||||
}
|
||||
return finish(p, plain, htmlText), nil
|
||||
}
|
||||
|
||||
func finish(p Parsed, plain, htmlText string) Parsed {
|
||||
p.Text = plain
|
||||
if strings.TrimSpace(p.Text) == "" {
|
||||
p.Text = htmlText
|
||||
}
|
||||
p.Text = strings.ToValidUTF8(p.Text, "�")
|
||||
if len(p.Text) > MostBodyText {
|
||||
p.Text = strings.ToValidUTF8(p.Text[:MostBodyText], "") + "\n[the text goes on; it was cut at 256 KiB]"
|
||||
}
|
||||
return p
|
||||
}
|
||||
|
||||
func orOctets(ct string) string {
|
||||
if ct == "" {
|
||||
return "application/octet-stream"
|
||||
}
|
||||
if t, _, err := mime.ParseMediaType(ct); err == nil {
|
||||
return t
|
||||
}
|
||||
return ct
|
||||
}
|
||||
|
||||
var (
|
||||
scriptsAndStyles = regexp.MustCompile(`(?is)<(script|style|head)\b.*?</(script|style|head)>`)
|
||||
breaks = regexp.MustCompile(`(?i)<(br|/p|/div|/tr|/li|/h[1-6])\b[^>]*>`)
|
||||
tags = regexp.MustCompile(`(?s)<[^>]*>`)
|
||||
blankLines = regexp.MustCompile(`\n[ \t]*\n([ \t]*\n)+`)
|
||||
)
|
||||
|
||||
// HTMLText is an HTML body as plain text: no script, style or markup, the entities read, paragraphs kept as lines.
|
||||
// Text only, so nothing of the sender's markup ever reaches whoever reads it.
|
||||
func HTMLText(s string) string {
|
||||
s = scriptsAndStyles.ReplaceAllString(s, "")
|
||||
s = breaks.ReplaceAllString(s, "\n")
|
||||
s = tags.ReplaceAllString(s, "")
|
||||
s = html.UnescapeString(s)
|
||||
s = strings.ReplaceAll(s, "\r", "")
|
||||
s = blankLines.ReplaceAllString(s, "\n\n")
|
||||
return strings.TrimSpace(s)
|
||||
}
|
||||
@@ -0,0 +1,275 @@
|
||||
package main
|
||||
|
||||
// The attachment store (novox/hq ADR 0359 §3): each attachment kept once, by its SHA-256, in the bucket the `minio`
|
||||
// module provides this module through the `s3-bucket` provision. The mesh writes the bucket's facts and its key into
|
||||
// one file (MESH_MAIL_STORE); this speaks S3 to it with the standard library alone: a HEAD, a PUT and a GET, each
|
||||
// signed with AWS Signature Version 4, path-style, as an S3-compatible object store answers them.
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"crypto/hmac"
|
||||
"crypto/sha256"
|
||||
"encoding/hex"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"net/http"
|
||||
"net/url"
|
||||
"os"
|
||||
"sort"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
)
|
||||
|
||||
// Objects is the attachment store.
|
||||
type Objects interface {
|
||||
// Bucket is the bucket's name, as an event names it.
|
||||
Bucket() string
|
||||
// Has says whether the key is kept.
|
||||
Has(ctx context.Context, key string) (bool, error)
|
||||
// Put keeps the bytes under the key.
|
||||
Put(ctx context.Context, key, mediaType string, data []byte) error
|
||||
// Get is the bytes kept under the key.
|
||||
Get(ctx context.Context, key string) ([]byte, error)
|
||||
}
|
||||
|
||||
// Keep stores an attachment once by its SHA-256 and answers its reference: the same file mailed ten times is one
|
||||
// object.
|
||||
func Keep(ctx context.Context, o Objects, data []byte, mediaType, name string) (AttachmentRef, error) {
|
||||
sum := sha256.Sum256(data)
|
||||
key := hex.EncodeToString(sum[:])
|
||||
has, err := o.Has(ctx, key)
|
||||
if err != nil {
|
||||
return AttachmentRef{}, err
|
||||
}
|
||||
if !has {
|
||||
if err := o.Put(ctx, key, mediaType, data); err != nil {
|
||||
return AttachmentRef{}, err
|
||||
}
|
||||
}
|
||||
return AttachmentRef{Bucket: o.Bucket(), Key: key, Size: int64(len(data)), Type: mediaType, Name: name}, nil
|
||||
}
|
||||
|
||||
// StoreFacts is the file the mesh writes from the s3-bucket binding and its secret.
|
||||
type StoreFacts struct {
|
||||
Scheme string `json:"scheme"`
|
||||
At string `json:"at"`
|
||||
Port string `json:"port"`
|
||||
Bucket string `json:"bucket"`
|
||||
Region string `json:"region"`
|
||||
Key string `json:"key"`
|
||||
Secret string `json:"secret"`
|
||||
}
|
||||
|
||||
// ReadStoreFacts reads the store's facts.
|
||||
func ReadStoreFacts(path string) (StoreFacts, error) {
|
||||
var f StoreFacts
|
||||
if path == "" {
|
||||
return f, errors.New("the module was started without its store's facts (MESH_MAIL_STORE)")
|
||||
}
|
||||
raw, err := os.ReadFile(path)
|
||||
if err != nil {
|
||||
return f, fmt.Errorf("the store's facts cannot be read: %w", err)
|
||||
}
|
||||
if err := json.Unmarshal(raw, &f); err != nil {
|
||||
return f, fmt.Errorf("the store's facts are not the module's: %v", err)
|
||||
}
|
||||
if f.At == "" || f.Bucket == "" || f.Key == "" || f.Secret == "" {
|
||||
return f, errors.New("the store's facts name no address, bucket or key")
|
||||
}
|
||||
if f.Scheme == "" {
|
||||
f.Scheme = "http"
|
||||
}
|
||||
if f.Region == "" {
|
||||
f.Region = "us-east-1"
|
||||
}
|
||||
return f, nil
|
||||
}
|
||||
|
||||
// S3 is the attachment store over S3.
|
||||
type S3 struct {
|
||||
Endpoint string // scheme://host[:port]
|
||||
Name string
|
||||
Region string
|
||||
Access string
|
||||
Secret string
|
||||
Client *http.Client
|
||||
Now func() time.Time
|
||||
}
|
||||
|
||||
// NewS3 is the store the facts name.
|
||||
func NewS3(f StoreFacts) *S3 {
|
||||
host := f.At
|
||||
if f.Port != "" {
|
||||
host += ":" + f.Port
|
||||
}
|
||||
return &S3{Endpoint: f.Scheme + "://" + host, Name: f.Bucket, Region: f.Region, Access: f.Key, Secret: f.Secret,
|
||||
Client: &http.Client{Timeout: 2 * time.Minute}, Now: time.Now}
|
||||
}
|
||||
|
||||
func (s *S3) Bucket() string { return s.Name }
|
||||
|
||||
func (s *S3) do(ctx context.Context, method, key, mediaType string, body []byte) (*http.Response, error) {
|
||||
u, err := url.Parse(s.Endpoint + "/" + s.Name + "/" + key)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
req, err := http.NewRequestWithContext(ctx, method, u.String(), bytes.NewReader(body))
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if mediaType != "" {
|
||||
req.Header.Set("Content-Type", mediaType)
|
||||
}
|
||||
req.ContentLength = int64(len(body))
|
||||
Sign(req, body, s.Access, s.Secret, s.Region, s.Now().UTC())
|
||||
return s.Client.Do(req)
|
||||
}
|
||||
|
||||
func (s *S3) Has(ctx context.Context, key string) (bool, error) {
|
||||
resp, err := s.do(ctx, http.MethodHead, key, "", nil)
|
||||
if err != nil {
|
||||
return false, fmt.Errorf("the attachment store cannot be reached: %w", err)
|
||||
}
|
||||
resp.Body.Close()
|
||||
switch resp.StatusCode {
|
||||
case http.StatusOK:
|
||||
return true, nil
|
||||
case http.StatusNotFound:
|
||||
return false, nil
|
||||
}
|
||||
return false, fmt.Errorf("the attachment store answered %s to asking for %s", resp.Status, key)
|
||||
}
|
||||
|
||||
func (s *S3) Put(ctx context.Context, key, mediaType string, data []byte) error {
|
||||
resp, err := s.do(ctx, http.MethodPut, key, mediaType, data)
|
||||
if err != nil {
|
||||
return fmt.Errorf("the attachment store cannot be reached: %w", err)
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
if resp.StatusCode/100 != 2 {
|
||||
said, _ := io.ReadAll(io.LimitReader(resp.Body, 512))
|
||||
return fmt.Errorf("the attachment store refused %s: %s %s", key, resp.Status, strings.TrimSpace(string(said)))
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *S3) Get(ctx context.Context, key string) ([]byte, error) {
|
||||
resp, err := s.do(ctx, http.MethodGet, key, "", nil)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("the attachment store cannot be reached: %w", err)
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
if resp.StatusCode != http.StatusOK {
|
||||
return nil, fmt.Errorf("the attachment store answered %s for %s", resp.Status, key)
|
||||
}
|
||||
return io.ReadAll(resp.Body)
|
||||
}
|
||||
|
||||
// Sign signs a request with AWS Signature Version 4 for S3: every header it carries but the agent and the length is
|
||||
// signed, with the payload's hash.
|
||||
func Sign(req *http.Request, body []byte, access, secret, region string, now time.Time) {
|
||||
payload := sha256.Sum256(body)
|
||||
payloadHex := hex.EncodeToString(payload[:])
|
||||
amzDate := now.Format("20060102T150405Z")
|
||||
day := now.Format("20060102")
|
||||
req.Header.Set("x-amz-date", amzDate)
|
||||
req.Header.Set("x-amz-content-sha256", payloadHex)
|
||||
|
||||
headers := map[string]string{"host": req.URL.Host}
|
||||
for k, v := range req.Header {
|
||||
lk := strings.ToLower(k)
|
||||
if lk == "user-agent" || lk == "content-length" || lk == "authorization" {
|
||||
continue
|
||||
}
|
||||
headers[lk] = strings.TrimSpace(strings.Join(v, ","))
|
||||
}
|
||||
names := make([]string, 0, len(headers))
|
||||
for k := range headers {
|
||||
names = append(names, k)
|
||||
}
|
||||
sort.Strings(names)
|
||||
var canonHeaders strings.Builder
|
||||
for _, k := range names {
|
||||
canonHeaders.WriteString(k + ":" + headers[k] + "\n")
|
||||
}
|
||||
signed := strings.Join(names, ";")
|
||||
path := req.URL.EscapedPath()
|
||||
if path == "" {
|
||||
path = "/"
|
||||
}
|
||||
canonical := strings.Join([]string{req.Method, path, canonicalQuery(req.URL.Query()), canonHeaders.String(), signed, payloadHex}, "\n")
|
||||
scope := day + "/" + region + "/s3/aws4_request"
|
||||
sum := sha256.Sum256([]byte(canonical))
|
||||
toSign := "AWS4-HMAC-SHA256\n" + amzDate + "\n" + scope + "\n" + hex.EncodeToString(sum[:])
|
||||
key := hmacOf([]byte("AWS4"+secret), day)
|
||||
key = hmacOf(key, region)
|
||||
key = hmacOf(key, "s3")
|
||||
key = hmacOf(key, "aws4_request")
|
||||
sig := hex.EncodeToString(hmacOf(key, toSign))
|
||||
req.Header.Set("Authorization", "AWS4-HMAC-SHA256 Credential="+access+"/"+scope+", SignedHeaders="+signed+", Signature="+sig)
|
||||
}
|
||||
|
||||
func canonicalQuery(q url.Values) string {
|
||||
keys := make([]string, 0, len(q))
|
||||
for k := range q {
|
||||
keys = append(keys, k)
|
||||
}
|
||||
sort.Strings(keys)
|
||||
var parts []string
|
||||
for _, k := range keys {
|
||||
vs := append([]string(nil), q[k]...)
|
||||
sort.Strings(vs)
|
||||
for _, v := range vs {
|
||||
parts = append(parts, url.QueryEscape(k)+"="+strings.ReplaceAll(url.QueryEscape(v), "+", "%20"))
|
||||
}
|
||||
}
|
||||
return strings.Join(parts, "&")
|
||||
}
|
||||
|
||||
func hmacOf(key []byte, data string) []byte {
|
||||
h := hmac.New(sha256.New, key)
|
||||
h.Write([]byte(data))
|
||||
return h.Sum(nil)
|
||||
}
|
||||
|
||||
// ---- in memory ----------------------------------------------------------------------------------------------------
|
||||
|
||||
type memObjects struct {
|
||||
mu sync.Mutex
|
||||
name string
|
||||
kept map[string][]byte
|
||||
puts int
|
||||
}
|
||||
|
||||
func newMemObjects() *memObjects { return &memObjects{name: "mail-test", kept: map[string][]byte{}} }
|
||||
|
||||
func (m *memObjects) Bucket() string { return m.name }
|
||||
|
||||
func (m *memObjects) Has(_ context.Context, key string) (bool, error) {
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
_, ok := m.kept[key]
|
||||
return ok, nil
|
||||
}
|
||||
|
||||
func (m *memObjects) Put(_ context.Context, key, _ string, data []byte) error {
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
m.kept[key] = append([]byte(nil), data...)
|
||||
m.puts++
|
||||
return nil
|
||||
}
|
||||
|
||||
func (m *memObjects) Get(_ context.Context, key string) ([]byte, error) {
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
b, ok := m.kept[key]
|
||||
if !ok {
|
||||
return nil, fmt.Errorf("%s is not kept", key)
|
||||
}
|
||||
return b, nil
|
||||
}
|
||||
@@ -0,0 +1,104 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"io"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
// The signature is AWS Signature Version 4 as S3 documents it: its own worked example, "GET Object", signs to the
|
||||
// signature AWS publishes.
|
||||
func TestTheSignatureIsTheOneS3Documents(t *testing.T) {
|
||||
req, _ := http.NewRequest(http.MethodGet, "https://examplebucket.s3.amazonaws.com/test.txt", nil)
|
||||
req.Header.Set("Range", "bytes=0-9")
|
||||
Sign(req, nil, "AKIAIOSFODNN7EXAMPLE", "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY", "us-east-1", time.Date(2013, 5, 24, 0, 0, 0, 0, time.UTC))
|
||||
want := "AWS4-HMAC-SHA256 Credential=AKIAIOSFODNN7EXAMPLE/20130524/us-east-1/s3/aws4_request, " +
|
||||
"SignedHeaders=host;range;x-amz-content-sha256;x-amz-date, Signature=f0e8bdb87c964420e857bd35b5d6ed310bd44f0170aba48dd91039c6036bdb41"
|
||||
if got := req.Header.Get("Authorization"); got != want {
|
||||
t.Fatalf("signed\n%s\nnot\n%s", got, want)
|
||||
}
|
||||
}
|
||||
|
||||
// An attachment is kept once by its SHA-256: the same file twice is one PUT, and it reads back.
|
||||
func TestAnAttachmentIsKeptOnceByItsHash(t *testing.T) {
|
||||
var mu sync.Mutex
|
||||
kept := map[string][]byte{}
|
||||
puts := 0
|
||||
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
if !strings.HasPrefix(r.Header.Get("Authorization"), "AWS4-HMAC-SHA256 Credential=mesh_novox_mail/") {
|
||||
w.WriteHeader(http.StatusForbidden)
|
||||
return
|
||||
}
|
||||
mu.Lock()
|
||||
defer mu.Unlock()
|
||||
switch r.Method {
|
||||
case http.MethodHead:
|
||||
if _, ok := kept[r.URL.Path]; !ok {
|
||||
w.WriteHeader(http.StatusNotFound)
|
||||
}
|
||||
case http.MethodPut:
|
||||
body, _ := io.ReadAll(r.Body)
|
||||
kept[r.URL.Path] = body
|
||||
puts++
|
||||
case http.MethodGet:
|
||||
b, ok := kept[r.URL.Path]
|
||||
if !ok {
|
||||
w.WriteHeader(http.StatusNotFound)
|
||||
return
|
||||
}
|
||||
_, _ = w.Write(b)
|
||||
}
|
||||
}))
|
||||
defer srv.Close()
|
||||
dir := t.TempDir()
|
||||
path := filepath.Join(dir, "store.json")
|
||||
host, port, _ := strings.Cut(strings.TrimPrefix(srv.URL, "http://"), ":")
|
||||
writeFile(t, path, `{"scheme": "http", "at": "`+host+`", "port": "`+port+`", "bucket": "mesh-novox-mail", "region": "eu-west",
|
||||
"key": "mesh_novox_mail", "secret": "s3cr3t"}`)
|
||||
facts, err := ReadStoreFacts(path)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
s3 := NewS3(facts)
|
||||
ctx := context.Background()
|
||||
a, err := Keep(ctx, s3, []byte("same file"), "text/plain", "a.txt")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
b, err := Keep(ctx, s3, []byte("same file"), "text/plain", "b.txt")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if a.Key != b.Key || a.Bucket != "mesh-novox-mail" || puts != 1 || len(a.Key) != 64 {
|
||||
t.Fatalf("%+v %+v, %d put(s)", a, b, puts)
|
||||
}
|
||||
if _, ok := kept["/mesh-novox-mail/"+a.Key]; !ok {
|
||||
t.Fatalf("kept %v", kept)
|
||||
}
|
||||
got, err := s3.Get(ctx, a.Key)
|
||||
if err != nil || string(got) != "same file" {
|
||||
t.Fatalf("%q %v", got, err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestAMessagesTextAndAttachmentsAreRead(t *testing.T) {
|
||||
p, err := ParseMessage([]byte(strings.ReplaceAll(attachedMail("x"), "\n", "\r\n")))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if strings.TrimSpace(p.Text) != "Please find the invoice attached." || len(p.Attachments) != 1 ||
|
||||
p.Attachments[0].Name != "invoice.pdf" || p.Attachments[0].Type != "application/pdf" || string(p.Attachments[0].Data) != attachmentData {
|
||||
t.Fatalf("%+v", p)
|
||||
}
|
||||
html := "From: a@b.c\r\nSubject: s\r\nContent-Type: text/html; charset=utf-8\r\n\r\n<html><head><style>p{}</style></head><body><p>Hello & welcome</p><script>alert(1)</script><p>Bye</p></body></html>"
|
||||
p, err = ParseMessage([]byte(html))
|
||||
if err != nil || p.Text != "Hello & welcome\nBye" {
|
||||
t.Fatalf("%q %v", p.Text, err)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,504 @@
|
||||
package main
|
||||
|
||||
// The index on the mesh's postgres (novox/hq ADR 0359 §3): the database the mesh provisions for this module through
|
||||
// the postgres-database provision, its address in the file DATABASE_URL_FILE names. Its schema is numbered migrations
|
||||
// in migrations/, applied by the preparation step (`mesh-mail prepare`, ADR 0135) and recorded in the database itself;
|
||||
// a fresh database is given the consolidated schema, schema.sql, at once, with every migration recorded as applied.
|
||||
// Nothing else changes the schema: the keeper, the watch and the tools refuse a schema that is behind, and say so.
|
||||
|
||||
import (
|
||||
"context"
|
||||
"crypto/sha256"
|
||||
"embed"
|
||||
"encoding/hex"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io/fs"
|
||||
"os"
|
||||
"path"
|
||||
"regexp"
|
||||
"sort"
|
||||
"strconv"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/jackc/pgx/v5"
|
||||
"github.com/jackc/pgx/v5/pgconn"
|
||||
"github.com/jackc/pgx/v5/pgxpool"
|
||||
)
|
||||
|
||||
//go:embed migrations/*.sql
|
||||
var migrationFiles embed.FS
|
||||
|
||||
//go:embed schema.sql
|
||||
var consolidated string
|
||||
|
||||
// Migration is one numbered change of the schema.
|
||||
type Migration struct {
|
||||
Number int
|
||||
Name string
|
||||
SQL string
|
||||
Checksum string
|
||||
}
|
||||
|
||||
var migrationFile = regexp.MustCompile(`^(\d{4})-([a-z0-9-]+)\.sql$`)
|
||||
|
||||
// Migrations are the migrations this binary carries, in order, numbered from 0001 without a gap.
|
||||
func Migrations() ([]Migration, error) {
|
||||
entries, err := fs.ReadDir(migrationFiles, "migrations")
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
var out []Migration
|
||||
for _, e := range entries {
|
||||
m := migrationFile.FindStringSubmatch(e.Name())
|
||||
if m == nil {
|
||||
return nil, fmt.Errorf("%s is not a migration's name (NNNN-what-it-does.sql)", e.Name())
|
||||
}
|
||||
n, _ := strconv.Atoi(m[1])
|
||||
body, err := fs.ReadFile(migrationFiles, path.Join("migrations", e.Name()))
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if strings.TrimSpace(string(body)) == "" {
|
||||
return nil, fmt.Errorf("%s is empty", e.Name())
|
||||
}
|
||||
sum := sha256.Sum256(body)
|
||||
out = append(out, Migration{Number: n, Name: m[2], SQL: string(body), Checksum: hex.EncodeToString(sum[:])})
|
||||
}
|
||||
sort.Slice(out, func(i, j int) bool { return out[i].Number < out[j].Number })
|
||||
for i, m := range out {
|
||||
if m.Number != i+1 {
|
||||
return nil, fmt.Errorf("the migrations are not numbered 0001 on without a gap: %04d is the %d. one", m.Number, i+1)
|
||||
}
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
const ledger = `create table if not exists migration (
|
||||
number integer primary key,
|
||||
name text not null,
|
||||
checksum text not null,
|
||||
applied timestamptz not null default now())`
|
||||
|
||||
// lockKey is the advisory lock every migration run takes, so two preparations never interleave.
|
||||
const lockKey int64 = 4_250_359_001
|
||||
|
||||
// PgStore is the Store on postgres.
|
||||
type PgStore struct{ pool *pgxpool.Pool }
|
||||
|
||||
// PgStoreFromEnv opens the index the mesh provisioned.
|
||||
func PgStoreFromEnv(ctx context.Context) (*PgStore, error) {
|
||||
file := os.Getenv("DATABASE_URL_FILE")
|
||||
if file == "" {
|
||||
return nil, errors.New("DATABASE_URL_FILE is not set: the index's database is a requirement the mesh resolves")
|
||||
}
|
||||
raw, err := os.ReadFile(file)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("the database's address cannot be read: %w", err)
|
||||
}
|
||||
return OpenPgStore(ctx, strings.TrimSpace(string(raw)))
|
||||
}
|
||||
|
||||
// OpenPgStore opens the index at an address.
|
||||
func OpenPgStore(ctx context.Context, url string) (*PgStore, error) {
|
||||
pool, err := pgxpool.New(ctx, url)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return &PgStore{pool: pool}, nil
|
||||
}
|
||||
|
||||
// Close closes the store.
|
||||
func (s *PgStore) Close() { s.pool.Close() }
|
||||
|
||||
// Prepare brings the schema to what this version needs, and says what it did.
|
||||
func (s *PgStore) Prepare(ctx context.Context) ([]string, error) {
|
||||
migrations, err := Migrations()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
conn, err := s.pool.Acquire(ctx)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer conn.Release()
|
||||
if _, err := conn.Exec(ctx, `select pg_advisory_lock($1)`, lockKey); err != nil {
|
||||
return nil, fmt.Errorf("the migration lock cannot be taken: %w", err)
|
||||
}
|
||||
defer func() { _, _ = conn.Exec(context.WithoutCancel(ctx), `select pg_advisory_unlock($1)`, lockKey) }()
|
||||
if _, err := conn.Exec(ctx, ledger); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
applied := map[int]string{}
|
||||
rows, err := conn.Query(ctx, `select number, checksum from migration`)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
highest := 0
|
||||
for rows.Next() {
|
||||
var n int
|
||||
var sum string
|
||||
if err := rows.Scan(&n, &sum); err != nil {
|
||||
rows.Close()
|
||||
return nil, err
|
||||
}
|
||||
applied[n] = sum
|
||||
highest = max(highest, n)
|
||||
}
|
||||
rows.Close()
|
||||
if err := rows.Err(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
record := func(tx pgx.Tx, m Migration) error {
|
||||
_, err := tx.Exec(ctx, `insert into migration (number, name, checksum) values ($1, $2, $3)`, m.Number, m.Name, m.Checksum)
|
||||
return err
|
||||
}
|
||||
if len(applied) == 0 {
|
||||
var exists bool
|
||||
if err := conn.QueryRow(ctx, `select to_regclass('message') is not null`).Scan(&exists); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if exists {
|
||||
return nil, errors.New("the database holds the index's tables and no record of a migration: its contents are not known, so nothing is changed")
|
||||
}
|
||||
tx, err := conn.Begin(ctx)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer func() { _ = tx.Rollback(context.WithoutCancel(ctx)) }()
|
||||
if _, err := tx.Exec(ctx, consolidated); err != nil {
|
||||
return nil, fmt.Errorf("the consolidated schema failed, and nothing of it was kept: %w", err)
|
||||
}
|
||||
for _, m := range migrations {
|
||||
if err := record(tx, m); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
if err := tx.Commit(ctx); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return []string{fmt.Sprintf("a fresh database: the consolidated schema, through migration %04d", len(migrations))}, nil
|
||||
}
|
||||
var problems []string
|
||||
var pending []Migration
|
||||
for _, m := range migrations {
|
||||
sum, ran := applied[m.Number]
|
||||
switch {
|
||||
case ran && sum != m.Checksum:
|
||||
problems = append(problems, fmt.Sprintf("migration %04d-%s ran, and its file has changed since: write a new one", m.Number, m.Name))
|
||||
case !ran && m.Number < highest:
|
||||
problems = append(problems, fmt.Sprintf("migration %04d-%s never ran, and %04d did: renumber it above %04d", m.Number, m.Name, highest, highest))
|
||||
case !ran:
|
||||
pending = append(pending, m)
|
||||
}
|
||||
}
|
||||
if len(problems) > 0 {
|
||||
return nil, errors.New("this database and these migrations disagree: " + strings.Join(problems, "; "))
|
||||
}
|
||||
var said []string
|
||||
for _, m := range pending {
|
||||
err := func() error {
|
||||
tx, err := conn.Begin(ctx)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer func() { _ = tx.Rollback(context.WithoutCancel(ctx)) }()
|
||||
if _, err := tx.Exec(ctx, m.SQL); err != nil {
|
||||
return fmt.Errorf("migration %04d-%s failed, and nothing it did was kept: %w", m.Number, m.Name, err)
|
||||
}
|
||||
if err := record(tx, m); err != nil {
|
||||
return err
|
||||
}
|
||||
return tx.Commit(ctx)
|
||||
}()
|
||||
if err != nil {
|
||||
return said, err
|
||||
}
|
||||
said = append(said, fmt.Sprintf("migration %04d-%s applied", m.Number, m.Name))
|
||||
}
|
||||
if len(said) == 0 {
|
||||
said = []string{fmt.Sprintf("the schema is at migration %04d, as this version needs", len(migrations))}
|
||||
}
|
||||
return said, nil
|
||||
}
|
||||
|
||||
// Ready says why the index cannot serve this version, or nil.
|
||||
func (s *PgStore) Ready(ctx context.Context) error {
|
||||
migrations, err := Migrations()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
var at int
|
||||
err = s.pool.QueryRow(ctx, `select coalesce(max(number), 0) from migration`).Scan(&at)
|
||||
var pgErr *pgconn.PgError
|
||||
if errors.As(err, &pgErr) && pgErr.Code == "42P01" {
|
||||
at = 0
|
||||
} else if err != nil {
|
||||
return fmt.Errorf("the database cannot be reached: %w", err)
|
||||
}
|
||||
if at != len(migrations) {
|
||||
return fmt.Errorf("the database's schema is at migration %04d and this version needs %04d: the preparation step "+
|
||||
"(mesh-mail prepare) has not run", at, len(migrations))
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *PgStore) Folder(ctx context.Context, account, folder string) (Folder, bool, error) {
|
||||
f := Folder{Account: account, Name: folder}
|
||||
var validity int64
|
||||
err := s.pool.QueryRow(ctx, `select uidvalidity, messages, synced from folder where account = $1 and name = $2`, account, folder).
|
||||
Scan(&validity, &f.Messages, &f.Synced)
|
||||
if errors.Is(err, pgx.ErrNoRows) {
|
||||
return f, false, nil
|
||||
}
|
||||
f.UIDValidity = uint32(validity)
|
||||
return f, err == nil, err
|
||||
}
|
||||
|
||||
func (s *PgStore) SetFolder(ctx context.Context, f Folder) error {
|
||||
_, err := s.pool.Exec(ctx, `insert into folder (account, name, uidvalidity, messages, synced) values ($1, $2, $3, $4, $5)
|
||||
on conflict (account, name) do update set uidvalidity = excluded.uidvalidity, messages = excluded.messages, synced = excluded.synced`,
|
||||
f.Account, f.Name, int64(f.UIDValidity), f.Messages, f.Synced)
|
||||
return err
|
||||
}
|
||||
|
||||
func (s *PgStore) DropFolder(ctx context.Context, account, folder string) error {
|
||||
return pgx.BeginFunc(ctx, s.pool, func(tx pgx.Tx) error {
|
||||
if _, err := tx.Exec(ctx, `delete from message where account = $1 and folder = $2`, account, folder); err != nil {
|
||||
return err
|
||||
}
|
||||
_, err := tx.Exec(ctx, `delete from folder where account = $1 and name = $2`, account, folder)
|
||||
return err
|
||||
})
|
||||
}
|
||||
|
||||
func (s *PgStore) Flags(ctx context.Context, account, folder string) (map[uint32][]string, error) {
|
||||
rows, err := s.pool.Query(ctx, `select uid, flags from message where account = $1 and folder = $2`, account, folder)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer rows.Close()
|
||||
out := map[uint32][]string{}
|
||||
for rows.Next() {
|
||||
var uid int64
|
||||
var flags []string
|
||||
if err := rows.Scan(&uid, &flags); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
out[uint32(uid)] = flags
|
||||
}
|
||||
return out, rows.Err()
|
||||
}
|
||||
|
||||
func keepEvent(ctx context.Context, tx pgx.Tx, e *Event) error {
|
||||
if e == nil {
|
||||
return nil
|
||||
}
|
||||
body, err := json.Marshal(e.Body)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
_, err = tx.Exec(ctx, `insert into outbox (key, body) values ($1, $2)`, e.Key, body)
|
||||
return err
|
||||
}
|
||||
|
||||
func (s *PgStore) Add(ctx context.Context, m Message, e *Event) error {
|
||||
from, _ := json.Marshal(m.From)
|
||||
to, _ := json.Marshal(m.To)
|
||||
cc, _ := json.Marshal(orNone(m.Cc))
|
||||
atts, _ := json.Marshal(m.Attachments)
|
||||
flags := m.Flags
|
||||
if flags == nil {
|
||||
flags = []string{}
|
||||
}
|
||||
people := strings.Join([]string{peopleText(m.From), peopleText(m.To), peopleText(m.Cc)}, " ")
|
||||
// An address is one word to the search, so its parts are kept beside it: "bob" finds bob@example.org.
|
||||
people += " " + strings.NewReplacer("@", " ", ".", " ", ",", " ").Replace(people)
|
||||
return pgx.BeginFunc(ctx, s.pool, func(tx pgx.Tx) error {
|
||||
if _, err := tx.Exec(ctx, `insert into message (account, folder, uid, uidvalidity, message_id, in_reply_to, date, sender,
|
||||
recipients, copied, people, subject, size, flags, attachments, body)
|
||||
values ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16)
|
||||
on conflict (account, folder, uid) do update set uidvalidity = excluded.uidvalidity, message_id = excluded.message_id,
|
||||
in_reply_to = excluded.in_reply_to, date = excluded.date, sender = excluded.sender, recipients = excluded.recipients,
|
||||
copied = excluded.copied, people = excluded.people, subject = excluded.subject, size = excluded.size,
|
||||
flags = excluded.flags, attachments = excluded.attachments, body = excluded.body`,
|
||||
m.Account, m.Folder, int64(m.UID), int64(m.UIDValidity), m.MessageID, m.InReplyTo, m.Date, from, to, cc, people,
|
||||
strings.ToValidUTF8(strings.ReplaceAll(m.Subject, "\x00", ""), ""), m.Size, flags, atts,
|
||||
strings.ReplaceAll(m.Body, "\x00", "")); err != nil {
|
||||
return err
|
||||
}
|
||||
return keepEvent(ctx, tx, e)
|
||||
})
|
||||
}
|
||||
|
||||
func orNone(ps []Person) []Person {
|
||||
if ps == nil {
|
||||
return []Person{}
|
||||
}
|
||||
return ps
|
||||
}
|
||||
|
||||
func (s *PgStore) SetFlags(ctx context.Context, account, folder string, uid uint32, flags []string, e *Event) error {
|
||||
return pgx.BeginFunc(ctx, s.pool, func(tx pgx.Tx) error {
|
||||
if _, err := tx.Exec(ctx, `update message set flags = $4 where account = $1 and folder = $2 and uid = $3`,
|
||||
account, folder, int64(uid), flags); err != nil {
|
||||
return err
|
||||
}
|
||||
return keepEvent(ctx, tx, e)
|
||||
})
|
||||
}
|
||||
|
||||
func (s *PgStore) Remove(ctx context.Context, account, folder string, uid uint32, e *Event) error {
|
||||
return pgx.BeginFunc(ctx, s.pool, func(tx pgx.Tx) error {
|
||||
if _, err := tx.Exec(ctx, `delete from message where account = $1 and folder = $2 and uid = $3`, account, folder, int64(uid)); err != nil {
|
||||
return err
|
||||
}
|
||||
return keepEvent(ctx, tx, e)
|
||||
})
|
||||
}
|
||||
|
||||
const messageColumns = `account, folder, uid, uidvalidity, message_id, in_reply_to, date, sender, recipients, copied, subject,
|
||||
size, flags, attachments`
|
||||
|
||||
func scanMessage(row pgx.Row, withBody bool) (Message, error) {
|
||||
var m Message
|
||||
var uid, validity int64
|
||||
var from, to, cc, atts []byte
|
||||
dest := []any{&m.Account, &m.Folder, &uid, &validity, &m.MessageID, &m.InReplyTo, &m.Date, &from, &to, &cc, &m.Subject,
|
||||
&m.Size, &m.Flags, &atts}
|
||||
if withBody {
|
||||
dest = append(dest, &m.Body)
|
||||
}
|
||||
if err := row.Scan(dest...); err != nil {
|
||||
return m, err
|
||||
}
|
||||
m.UID, m.UIDValidity, m.Date = uint32(uid), uint32(validity), m.Date.UTC()
|
||||
for _, f := range []struct {
|
||||
raw []byte
|
||||
into any
|
||||
}{{from, &m.From}, {to, &m.To}, {cc, &m.Cc}, {atts, &m.Attachments}} {
|
||||
if err := json.Unmarshal(f.raw, f.into); err != nil {
|
||||
return m, err
|
||||
}
|
||||
}
|
||||
return m, nil
|
||||
}
|
||||
|
||||
func (s *PgStore) Get(ctx context.Context, account, folder string, uid uint32) (Message, bool, error) {
|
||||
m, err := scanMessage(s.pool.QueryRow(ctx, `select `+messageColumns+`, body from message where account = $1 and folder = $2 and uid = $3`,
|
||||
account, folder, int64(uid)), true)
|
||||
if errors.Is(err, pgx.ErrNoRows) {
|
||||
return Message{}, false, nil
|
||||
}
|
||||
return m, err == nil, err
|
||||
}
|
||||
|
||||
func (s *PgStore) Search(ctx context.Context, q Query) ([]Message, error) {
|
||||
limit := q.Limit
|
||||
if limit <= 0 {
|
||||
limit = 20
|
||||
}
|
||||
rows, err := s.pool.Query(ctx, `select `+messageColumns+` from message
|
||||
where account = any($1) and ($2 = '' or folder = $2) and search @@ plainto_tsquery('simple'::regconfig, $3)
|
||||
order by date desc limit $4`, q.Accounts, q.Folder, q.Text, limit)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer rows.Close()
|
||||
var out []Message
|
||||
for rows.Next() {
|
||||
m, err := scanMessage(rows, false)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
out = append(out, m)
|
||||
}
|
||||
return out, rows.Err()
|
||||
}
|
||||
|
||||
func (s *PgStore) Folders(ctx context.Context, account string) ([]Folder, error) {
|
||||
rows, err := s.pool.Query(ctx, `select name, uidvalidity, messages, synced from folder where account = $1 order by name`, account)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer rows.Close()
|
||||
var out []Folder
|
||||
for rows.Next() {
|
||||
f := Folder{Account: account}
|
||||
var validity int64
|
||||
if err := rows.Scan(&f.Name, &validity, &f.Messages, &f.Synced); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
f.UIDValidity, f.Synced = uint32(validity), f.Synced.UTC()
|
||||
out = append(out, f)
|
||||
}
|
||||
return out, rows.Err()
|
||||
}
|
||||
|
||||
func (s *PgStore) Pending(ctx context.Context, limit int) ([]Event, error) {
|
||||
rows, err := s.pool.Query(ctx, `select id, key, body from outbox order by id limit $1`, limit)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer rows.Close()
|
||||
var out []Event
|
||||
for rows.Next() {
|
||||
var e Event
|
||||
var body []byte
|
||||
if err := rows.Scan(&e.ID, &e.Key, &body); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if err := json.Unmarshal(body, &e.Body); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
out = append(out, e)
|
||||
}
|
||||
return out, rows.Err()
|
||||
}
|
||||
|
||||
func (s *PgStore) Said(ctx context.Context, ids []int64) error {
|
||||
_, err := s.pool.Exec(ctx, `delete from outbox where id = any($1)`, ids)
|
||||
return err
|
||||
}
|
||||
|
||||
func (s *PgStore) SetState(ctx context.Context, st AccountState) error {
|
||||
_, err := s.pool.Exec(ctx, `insert into account_state (account, state, why, heard, synced, said) values ($1, $2, $3, $4, $5, $6)
|
||||
on conflict (account) do update set state = excluded.state, why = excluded.why, heard = excluded.heard,
|
||||
synced = excluded.synced, said = excluded.said`,
|
||||
st.Account, st.State, st.Why, nullTime(st.Heard), nullTime(st.Synced), st.Said)
|
||||
return err
|
||||
}
|
||||
|
||||
func nullTime(t time.Time) *time.Time {
|
||||
if t.IsZero() {
|
||||
return nil
|
||||
}
|
||||
return &t
|
||||
}
|
||||
|
||||
func (s *PgStore) States(ctx context.Context) (map[string]AccountState, error) {
|
||||
rows, err := s.pool.Query(ctx, `select account, state, why, heard, synced, said from account_state`)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer rows.Close()
|
||||
out := map[string]AccountState{}
|
||||
for rows.Next() {
|
||||
var st AccountState
|
||||
var heard, synced *time.Time
|
||||
if err := rows.Scan(&st.Account, &st.State, &st.Why, &heard, &synced, &st.Said); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if heard != nil {
|
||||
st.Heard = heard.UTC()
|
||||
}
|
||||
if synced != nil {
|
||||
st.Synced = synced.UTC()
|
||||
}
|
||||
st.Said = st.Said.UTC()
|
||||
out[st.Account] = st
|
||||
}
|
||||
return out, rows.Err()
|
||||
}
|
||||
@@ -0,0 +1,99 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"reflect"
|
||||
"strings"
|
||||
"testing"
|
||||
)
|
||||
|
||||
// The schema is numbered migrations and a consolidated schema for a fresh database (the mesh's rule: both). A
|
||||
// database given the consolidated schema and one given every migration in order hold the same tables and columns.
|
||||
func TestTheConsolidatedSchemaIsTheMigrationsApplied(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
fresh := pgStore(t)
|
||||
stepped, err := OpenPgStore(ctx, freshDatabase(t))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer stepped.Close()
|
||||
migrations, err := Migrations()
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
for _, m := range migrations {
|
||||
if _, err := stepped.pool.Exec(ctx, m.SQL); err != nil {
|
||||
t.Fatalf("migration %04d: %v", m.Number, err)
|
||||
}
|
||||
}
|
||||
a, b := shape(t, fresh), shape(t, stepped)
|
||||
if !reflect.DeepEqual(a, b) || len(a) < 25 {
|
||||
t.Fatalf("the consolidated schema and the migrations differ:\nconsolidated %v\nmigrations %v", a, b)
|
||||
}
|
||||
if !strings.Contains(consolidated, "through migration 0001") || len(migrations) != 1 {
|
||||
t.Fatal("the consolidated schema does not say how far it is consolidated")
|
||||
}
|
||||
// Prepared twice, it changes nothing; and it is ready.
|
||||
if said, err := fresh.Prepare(ctx); err != nil || !strings.Contains(said[0], "as this version needs") {
|
||||
t.Fatalf("%v %v", said, err)
|
||||
}
|
||||
if err := fresh.Ready(ctx); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := stepped.Ready(ctx); err == nil {
|
||||
t.Fatal("a database whose migrations are not recorded is ready")
|
||||
}
|
||||
}
|
||||
|
||||
func shape(t *testing.T, s *PgStore) []string {
|
||||
t.Helper()
|
||||
rows, err := s.pool.Query(context.Background(), `select table_name || '.' || column_name || ' ' || data_type
|
||||
from information_schema.columns where table_schema = 'public' and table_name <> 'migration' order by 1`)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer rows.Close()
|
||||
var out []string
|
||||
for rows.Next() {
|
||||
var c string
|
||||
_ = rows.Scan(&c)
|
||||
out = append(out, c)
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// The search finds by every word, across the accounts asked only, newest first, and answers no body.
|
||||
func TestTheIndexSearchesByWordAndAccount(t *testing.T) {
|
||||
eachStore(t, func(t *testing.T, st Store) {
|
||||
ctx := context.Background()
|
||||
for i, subj := range []string{"the invoice of october", "lunch", "another invoice"} {
|
||||
m := Message{Account: "mesh", Folder: "INBOX", UID: uint32(i + 1), Subject: subj, Date: judgedAt.AddDate(0, 0, i),
|
||||
From: []Person{{Address: "bob@example.org"}}, To: []Person{}, Flags: []string{}, Attachments: []AttachmentRef{}, Body: "text " + subj}
|
||||
if err := st.Add(ctx, m, nil); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
other := Message{Account: "gmail", Folder: "INBOX", UID: 1, Subject: "invoice elsewhere", Date: judgedAt, From: []Person{}, To: []Person{},
|
||||
Flags: []string{}, Attachments: []AttachmentRef{}}
|
||||
if err := st.Add(ctx, other, nil); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
hits, err := st.Search(ctx, Query{Text: "invoice", Accounts: []string{"mesh"}, Limit: 10})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if len(hits) != 2 || hits[0].Subject != "another invoice" || hits[0].Body != "" {
|
||||
t.Fatalf("%+v", hits)
|
||||
}
|
||||
if hits, _ := st.Search(ctx, Query{Text: "invoice october", Accounts: []string{"mesh"}}); len(hits) != 1 {
|
||||
t.Fatalf("%+v", hits)
|
||||
}
|
||||
if hits, _ := st.Search(ctx, Query{Text: "bob", Accounts: []string{"mesh"}}); len(hits) != 3 {
|
||||
t.Fatalf("by sender: %+v", hits)
|
||||
}
|
||||
got, ok, err := st.Get(ctx, "mesh", "INBOX", 2)
|
||||
if err != nil || !ok || got.Body != "text lunch" {
|
||||
t.Fatalf("%+v %v %v", got, ok, err)
|
||||
}
|
||||
})
|
||||
}
|
||||
@@ -0,0 +1,46 @@
|
||||
-- The consolidated schema, through migration 0001: what a fresh database is given at once (the mesh's rule: a
|
||||
-- numbered migration for the running store, and this for a fresh install).
|
||||
create table folder (
|
||||
account text not null,
|
||||
name text not null,
|
||||
uidvalidity bigint not null,
|
||||
messages integer not null default 0,
|
||||
synced timestamptz not null default now(),
|
||||
primary key (account, name));
|
||||
|
||||
create table message (
|
||||
account text not null,
|
||||
folder text not null,
|
||||
uid bigint not null,
|
||||
uidvalidity bigint not null,
|
||||
message_id text not null default '',
|
||||
in_reply_to text not null default '',
|
||||
date timestamptz not null,
|
||||
sender jsonb not null default '[]',
|
||||
recipients jsonb not null default '[]',
|
||||
copied jsonb not null default '[]',
|
||||
people text not null default '',
|
||||
subject text not null default '',
|
||||
size bigint not null default 0,
|
||||
flags text[] not null default '{}',
|
||||
attachments jsonb not null default '[]',
|
||||
body text not null default '',
|
||||
search tsvector generated always as (
|
||||
to_tsvector('simple'::regconfig, subject || ' ' || people || ' ' || body)) stored,
|
||||
primary key (account, folder, uid));
|
||||
create index message_search on message using gin (search);
|
||||
create index message_date on message (account, date desc);
|
||||
|
||||
create table account_state (
|
||||
account text primary key,
|
||||
state text not null,
|
||||
why text not null default '',
|
||||
heard timestamptz,
|
||||
synced timestamptz,
|
||||
said timestamptz not null default now());
|
||||
|
||||
create table outbox (
|
||||
id bigserial primary key,
|
||||
key text not null,
|
||||
body jsonb not null,
|
||||
made timestamptz not null default now());
|
||||
@@ -0,0 +1,332 @@
|
||||
package main
|
||||
|
||||
// The index (novox/hq ADR 0359 §3): one row per message and folder, in the module's own database; the events still to
|
||||
// be said on the bus, written in the same step as the change they say; and each account's state as its keeper last
|
||||
// saw it, which the health watch reads. The Store is that, over postgres (pgstore.go) or in memory for the tests.
|
||||
|
||||
import (
|
||||
"context"
|
||||
"sort"
|
||||
"strconv"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
)
|
||||
|
||||
// Person is a sender or a recipient.
|
||||
type Person struct {
|
||||
Name string `json:"name,omitempty"`
|
||||
Address string `json:"address"`
|
||||
}
|
||||
|
||||
// AttachmentRef names an attachment by its content (§3): the bucket and key it is kept under, the key being its
|
||||
// SHA-256 in hex, with its size, media type and file name.
|
||||
type AttachmentRef struct {
|
||||
Bucket string `json:"bucket"`
|
||||
Key string `json:"key"`
|
||||
Size int64 `json:"size"`
|
||||
Type string `json:"type"`
|
||||
Name string `json:"name,omitempty"`
|
||||
}
|
||||
|
||||
// Message is one message in one folder of one account, as indexed.
|
||||
type Message struct {
|
||||
Account string `json:"account"`
|
||||
Folder string `json:"folder"`
|
||||
UIDValidity uint32 `json:"uidvalidity"`
|
||||
UID uint32 `json:"uid"`
|
||||
MessageID string `json:"message-id,omitempty"`
|
||||
InReplyTo string `json:"in-reply-to,omitempty"`
|
||||
Date time.Time `json:"date"`
|
||||
From []Person `json:"from"`
|
||||
To []Person `json:"to"`
|
||||
Cc []Person `json:"cc,omitempty"`
|
||||
Subject string `json:"subject"`
|
||||
Size int64 `json:"size"`
|
||||
Flags []string `json:"flags"`
|
||||
Attachments []AttachmentRef `json:"attachments"`
|
||||
// Body is the message's text, kept for search and served only through mail_read; never on the bus.
|
||||
Body string `json:"-"`
|
||||
}
|
||||
|
||||
// Event is one change said on the bus: its key (`mail.<account>.message.<added|changed|removed>`) and its body,
|
||||
// metadata only.
|
||||
type Event struct {
|
||||
ID int64
|
||||
Key string
|
||||
Body map[string]any
|
||||
}
|
||||
|
||||
// Folder is what the index knows of one folder.
|
||||
type Folder struct {
|
||||
Account string `json:"-"`
|
||||
Name string `json:"name"`
|
||||
UIDValidity uint32 `json:"-"`
|
||||
Messages int `json:"messages"`
|
||||
Synced time.Time `json:"synced"`
|
||||
}
|
||||
|
||||
// The states an account's keeper says it is in.
|
||||
const (
|
||||
AccountSyncing = "syncing"
|
||||
AccountCurrent = "current"
|
||||
AccountWaits = "waits-for-sign-in"
|
||||
AccountFailing = "failing"
|
||||
)
|
||||
|
||||
// AccountState is an account as its keeper last said it: its state, why, and when the server last answered.
|
||||
type AccountState struct {
|
||||
Account string `json:"account"`
|
||||
State string `json:"state"`
|
||||
Why string `json:"why"`
|
||||
Heard time.Time `json:"heard,omitempty"`
|
||||
Synced time.Time `json:"synced,omitempty"`
|
||||
Said time.Time `json:"said"`
|
||||
}
|
||||
|
||||
// Query is a search of the index.
|
||||
type Query struct {
|
||||
Text string
|
||||
Accounts []string
|
||||
Folder string
|
||||
Limit int
|
||||
}
|
||||
|
||||
// Store is the index.
|
||||
type Store interface {
|
||||
Ready(ctx context.Context) error
|
||||
// Folder is what the index knows of a folder; false when it never synced it.
|
||||
Folder(ctx context.Context, account, folder string) (Folder, bool, error)
|
||||
// SetFolder records a folder synced.
|
||||
SetFolder(ctx context.Context, f Folder) error
|
||||
// DropFolder forgets a folder's rows, without an event: the server renumbered it (a new UIDVALIDITY) and the
|
||||
// next sync indexes it afresh.
|
||||
DropFolder(ctx context.Context, account, folder string) error
|
||||
// Flags is every indexed message of a folder, by UID, with its flags.
|
||||
Flags(ctx context.Context, account, folder string) (map[uint32][]string, error)
|
||||
// Add indexes a message, and keeps the event that says it when there is one, in one step.
|
||||
Add(ctx context.Context, m Message, e *Event) error
|
||||
// SetFlags changes a message's flags, and keeps the event, in one step.
|
||||
SetFlags(ctx context.Context, account, folder string, uid uint32, flags []string, e *Event) error
|
||||
// Remove removes a message, and keeps the event, in one step.
|
||||
Remove(ctx context.Context, account, folder string, uid uint32, e *Event) error
|
||||
// Get is one message, with its body.
|
||||
Get(ctx context.Context, account, folder string, uid uint32) (Message, bool, error)
|
||||
// Search finds messages, newest first, without their bodies.
|
||||
Search(ctx context.Context, q Query) ([]Message, error)
|
||||
// Folders is every folder the index knows of an account.
|
||||
Folders(ctx context.Context, account string) ([]Folder, error)
|
||||
// Pending is the events not yet said, oldest first; Said removes those said.
|
||||
Pending(ctx context.Context, limit int) ([]Event, error)
|
||||
Said(ctx context.Context, ids []int64) error
|
||||
// SetState and States are each account's state, as its keeper says it.
|
||||
SetState(ctx context.Context, s AccountState) error
|
||||
States(ctx context.Context) (map[string]AccountState, error)
|
||||
}
|
||||
|
||||
// ---- in memory ----------------------------------------------------------------------------------------------------
|
||||
|
||||
type memStore struct {
|
||||
mu sync.Mutex
|
||||
messages map[string]Message
|
||||
folders map[string]Folder
|
||||
events []Event
|
||||
next int64
|
||||
states map[string]AccountState
|
||||
}
|
||||
|
||||
func newMemStore() *memStore {
|
||||
return &memStore{messages: map[string]Message{}, folders: map[string]Folder{}, states: map[string]AccountState{}}
|
||||
}
|
||||
|
||||
func msgKey(account, folder string, uid uint32) string {
|
||||
return account + "\x00" + folder + "\x00" + itoa(int64(uid))
|
||||
}
|
||||
|
||||
func folderKey(account, folder string) string { return account + "\x00" + folder }
|
||||
|
||||
func (s *memStore) Ready(context.Context) error { return nil }
|
||||
|
||||
func (s *memStore) Folder(_ context.Context, account, folder string) (Folder, bool, error) {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
f, ok := s.folders[folderKey(account, folder)]
|
||||
return f, ok, nil
|
||||
}
|
||||
|
||||
func (s *memStore) SetFolder(_ context.Context, f Folder) error {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
s.folders[folderKey(f.Account, f.Name)] = f
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *memStore) DropFolder(_ context.Context, account, folder string) error {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
for k, m := range s.messages {
|
||||
if m.Account == account && m.Folder == folder {
|
||||
delete(s.messages, k)
|
||||
}
|
||||
}
|
||||
delete(s.folders, folderKey(account, folder))
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *memStore) Flags(_ context.Context, account, folder string) (map[uint32][]string, error) {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
out := map[uint32][]string{}
|
||||
for _, m := range s.messages {
|
||||
if m.Account == account && m.Folder == folder {
|
||||
out[m.UID] = append([]string(nil), m.Flags...)
|
||||
}
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
func (s *memStore) keep(e *Event) {
|
||||
if e == nil {
|
||||
return
|
||||
}
|
||||
s.next++
|
||||
ev := *e
|
||||
ev.ID = s.next
|
||||
s.events = append(s.events, ev)
|
||||
}
|
||||
|
||||
func (s *memStore) Add(_ context.Context, m Message, e *Event) error {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
s.messages[msgKey(m.Account, m.Folder, m.UID)] = m
|
||||
s.keep(e)
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *memStore) SetFlags(_ context.Context, account, folder string, uid uint32, flags []string, e *Event) error {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
k := msgKey(account, folder, uid)
|
||||
if m, ok := s.messages[k]; ok {
|
||||
m.Flags = flags
|
||||
s.messages[k] = m
|
||||
}
|
||||
s.keep(e)
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *memStore) Remove(_ context.Context, account, folder string, uid uint32, e *Event) error {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
delete(s.messages, msgKey(account, folder, uid))
|
||||
s.keep(e)
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *memStore) Get(_ context.Context, account, folder string, uid uint32) (Message, bool, error) {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
m, ok := s.messages[msgKey(account, folder, uid)]
|
||||
return m, ok, nil
|
||||
}
|
||||
|
||||
func (s *memStore) Search(_ context.Context, q Query) ([]Message, error) {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
words := strings.Fields(strings.ToLower(q.Text))
|
||||
var out []Message
|
||||
for _, m := range s.messages {
|
||||
if !contains(q.Accounts, m.Account) || (q.Folder != "" && q.Folder != m.Folder) {
|
||||
continue
|
||||
}
|
||||
hay := strings.ToLower(m.Subject + " " + m.Body + " " + peopleText(m.From) + " " + peopleText(m.To) + " " + peopleText(m.Cc))
|
||||
all := true
|
||||
for _, w := range words {
|
||||
all = all && strings.Contains(hay, w)
|
||||
}
|
||||
if all {
|
||||
m.Body = ""
|
||||
out = append(out, m)
|
||||
}
|
||||
}
|
||||
sort.Slice(out, func(i, j int) bool { return out[i].Date.After(out[j].Date) })
|
||||
if q.Limit > 0 && len(out) > q.Limit {
|
||||
out = out[:q.Limit]
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
func (s *memStore) Folders(_ context.Context, account string) ([]Folder, error) {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
var out []Folder
|
||||
for _, f := range s.folders {
|
||||
if f.Account == account {
|
||||
out = append(out, f)
|
||||
}
|
||||
}
|
||||
sort.Slice(out, func(i, j int) bool { return out[i].Name < out[j].Name })
|
||||
return out, nil
|
||||
}
|
||||
|
||||
func (s *memStore) Pending(_ context.Context, limit int) ([]Event, error) {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
n := min(limit, len(s.events))
|
||||
return append([]Event(nil), s.events[:n]...), nil
|
||||
}
|
||||
|
||||
func (s *memStore) Said(_ context.Context, ids []int64) error {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
said := map[int64]bool{}
|
||||
for _, id := range ids {
|
||||
said[id] = true
|
||||
}
|
||||
kept := s.events[:0]
|
||||
for _, e := range s.events {
|
||||
if !said[e.ID] {
|
||||
kept = append(kept, e)
|
||||
}
|
||||
}
|
||||
s.events = kept
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *memStore) SetState(_ context.Context, st AccountState) error {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
s.states[st.Account] = st
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *memStore) States(context.Context) (map[string]AccountState, error) {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
out := map[string]AccountState{}
|
||||
for k, v := range s.states {
|
||||
out[k] = v
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
// ---- small helpers ------------------------------------------------------------------------------------------------
|
||||
|
||||
func contains(list []string, s string) bool {
|
||||
for _, x := range list {
|
||||
if x == s {
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
func peopleText(ps []Person) string {
|
||||
var parts []string
|
||||
for _, p := range ps {
|
||||
parts = append(parts, strings.TrimSpace(p.Name+" "+p.Address))
|
||||
}
|
||||
return strings.Join(parts, ", ")
|
||||
}
|
||||
|
||||
func itoa(n int64) string { return strconv.FormatInt(n, 10) }
|
||||
@@ -0,0 +1,340 @@
|
||||
package main
|
||||
|
||||
// The module's tools (novox/hq ADR 0359 §5): mail_accounts, mail_search, mail_read and mail_health. Who calls is the
|
||||
// runtime's word (stdio.CallerOf), never an argument; a caller reads an account's mail — its search, a body, an
|
||||
// attachment — only when the trusted setting `grants` names it for that account (§4). Nothing here writes to a mailbox.
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/base64"
|
||||
"errors"
|
||||
"fmt"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
stdio "git.novox.be/novox/mesh-sdk/go"
|
||||
)
|
||||
|
||||
// Tool names, as the manifest declares them.
|
||||
const (
|
||||
AccountsTool = "mail_accounts"
|
||||
SearchTool = "mail_search"
|
||||
ReadTool = "mail_read"
|
||||
)
|
||||
|
||||
// Bounds of what a tool answers.
|
||||
const (
|
||||
MostResults = 50
|
||||
MostQuery = 200
|
||||
MostBodyAnswered = 64 << 10
|
||||
MostAttachment = 8 << 20
|
||||
)
|
||||
|
||||
// Bundle is what the tools reach: the settings, the index and the attachment store.
|
||||
type Bundle struct {
|
||||
Env Env
|
||||
Index func(ctx context.Context) (Store, error)
|
||||
Objects func() (Objects, error)
|
||||
Caller func(args map[string]any) string
|
||||
Now func() time.Time
|
||||
}
|
||||
|
||||
func (b *Bundle) now() time.Time {
|
||||
if b.Now != nil {
|
||||
return b.Now()
|
||||
}
|
||||
return time.Now()
|
||||
}
|
||||
|
||||
// settings are the accounts and the grants. A grants setting that cannot be read grants nothing, and its reason is
|
||||
// kept, so a refusal says why rather than only that the caller is not granted.
|
||||
func (b *Bundle) settings() ([]Account, access, error) {
|
||||
accounts, err := ReadAccounts(b.Env.Accounts)
|
||||
if err != nil {
|
||||
return nil, access{}, err
|
||||
}
|
||||
grants, err := ReadGrants(b.Env.Grants)
|
||||
if err != nil {
|
||||
return accounts, access{grants: Grants{}, unread: err}, nil
|
||||
}
|
||||
return accounts, access{grants: grants}, nil
|
||||
}
|
||||
|
||||
// access is the grants as read, and why they could not be, when they could not.
|
||||
type access struct {
|
||||
grants Grants
|
||||
unread error
|
||||
}
|
||||
|
||||
func (a access) Allows(caller, account string) bool { return a.grants.Allows(caller, account) }
|
||||
|
||||
func (a access) Granted(caller string, accounts []Account) []string {
|
||||
return a.grants.Granted(caller, accounts)
|
||||
}
|
||||
|
||||
// refuse is the refusal of a caller for an account, saying why: the grants unread, or the caller not in them.
|
||||
func (a access) refuse(caller, account string) error {
|
||||
if a.unread != nil {
|
||||
return fmt.Errorf("refused: the setting grants cannot be read, so it grants nobody: %v", a.unread)
|
||||
}
|
||||
return refused(caller, account)
|
||||
}
|
||||
|
||||
// Tools are the module's tools.
|
||||
func (b *Bundle) Tools() []stdio.Tool {
|
||||
return []stdio.Tool{
|
||||
{Name: AccountsTool,
|
||||
Description: "The operator's mail accounts you are granted, each with its kind, its address, its state (syncing, " +
|
||||
"current, waits-for-sign-in or failing, and why), when its server last answered, when it was last brought up to " +
|
||||
"date, and its folders with how many messages each holds. An account you are not granted is not listed, and " +
|
||||
"nothing of it is said. How the module as a whole is, is mail_health's to say. Read-only.",
|
||||
Input: schema(nil, map[string]any{}),
|
||||
Run: b.accountsTool},
|
||||
{Name: SearchTool,
|
||||
Description: "Search the mail of the accounts you are granted: every word of the query must be in the subject, the " +
|
||||
"text, a sender or a recipient. Answers the newest first, at most 50: account, folder, uid, message id, date, " +
|
||||
"sender, recipients, subject, size, flags and attachment references, never a body. Refused for an account you " +
|
||||
"are not granted, and when you are granted none. Read-only.",
|
||||
Input: schema([]string{"query"}, map[string]any{
|
||||
"query": text("the words to find, at most 200 characters"),
|
||||
"account": text("one account by its name in mail_accounts; every account you are granted when left out"),
|
||||
"folder": text("one folder, such as INBOX; every folder when left out"),
|
||||
"limit": integer("how many to answer, at most 50 (20 when left out)"),
|
||||
}),
|
||||
Run: b.searchTool},
|
||||
{Name: ReadTool,
|
||||
Description: "Read one message of an account you are granted, as an event or mail_search names it (account, folder, " +
|
||||
"uid): its metadata and its text (the plain part, or the HTML part as text, at most 64 KiB here). With " +
|
||||
"attachment (its key, the SHA-256 the message's references name), that one attachment's bytes, base64, at most " +
|
||||
"8 MiB. Refused for an account you are not granted. Never marks a message seen. Read-only.",
|
||||
Input: schema([]string{"account", "folder", "uid"}, map[string]any{
|
||||
"account": text("the account, by its name in mail_accounts"),
|
||||
"folder": text("the folder, such as INBOX"),
|
||||
"uid": integer("the message's uid in that folder"),
|
||||
"attachment": text("an attachment's key, the SHA-256 the message's references name, to answer its bytes"),
|
||||
}),
|
||||
Run: b.readTool},
|
||||
{Name: HealthTool,
|
||||
Description: "How mesh-mail is, as its watch last saw it: healthy when every account's server answered within five " +
|
||||
"minutes; unhealthy, naming the account and why, when a connection is down or its IDLE on the inbox went silent; " +
|
||||
"waits, naming the account's password, while the operator has not given it at the desk.",
|
||||
Input: schema(nil, map[string]any{}),
|
||||
Run: func(map[string]any) (any, error) { return ReadStatus(b.Env.Status, b.now()), nil }},
|
||||
}
|
||||
}
|
||||
|
||||
func (b *Bundle) caller(args map[string]any) string {
|
||||
if b.Caller != nil {
|
||||
return b.Caller(args)
|
||||
}
|
||||
return stdio.CallerOf(args)
|
||||
}
|
||||
|
||||
func (b *Bundle) accountsTool(args map[string]any) (any, error) {
|
||||
if err := only(args); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
accounts, grants, err := b.settings()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second)
|
||||
defer cancel()
|
||||
st, err := b.Index(ctx)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("the index cannot be reached: %v", err)
|
||||
}
|
||||
states, err := st.States(ctx)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
caller := b.caller(args)
|
||||
out := []map[string]any{}
|
||||
for _, a := range accounts {
|
||||
// Only the accounts the caller is granted: of any other it learns nothing, not even that it exists.
|
||||
if !grants.Allows(caller, a.Name) {
|
||||
continue
|
||||
}
|
||||
entry := map[string]any{"account": a.Name, "kind": a.Kind, "address": a.Address}
|
||||
if s, ok := states[a.Name]; ok {
|
||||
entry["state"], entry["why"] = s.State, s.Why
|
||||
if !s.Heard.IsZero() {
|
||||
entry["heard"] = s.Heard.UTC().Format(time.RFC3339)
|
||||
}
|
||||
if !s.Synced.IsZero() {
|
||||
entry["synced"] = s.Synced.UTC().Format(time.RFC3339)
|
||||
}
|
||||
} else {
|
||||
entry["state"], entry["why"] = "unknown", "its keeper has said nothing of it yet"
|
||||
}
|
||||
folders, err := st.Folders(ctx, a.Name)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
entry["folders"] = folders
|
||||
out = append(out, entry)
|
||||
}
|
||||
answer := map[string]any{"accounts": out, "caller": caller}
|
||||
if grants.unread != nil {
|
||||
answer["why"] = "the setting grants cannot be read, so it grants nobody: " + grants.unread.Error()
|
||||
}
|
||||
return answer, nil
|
||||
}
|
||||
|
||||
func (b *Bundle) searchTool(args map[string]any) (any, error) {
|
||||
if err := only(args, "query", "account", "folder", "limit"); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
query, _ := args["query"].(string)
|
||||
if strings.TrimSpace(query) == "" || len(query) > MostQuery {
|
||||
return nil, fmt.Errorf("%s needs a query of 1 to %d characters", SearchTool, MostQuery)
|
||||
}
|
||||
account, _ := args["account"].(string)
|
||||
folder, _ := args["folder"].(string)
|
||||
limit, err := number(args, "limit", 20)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if limit < 1 || limit > MostResults {
|
||||
return nil, fmt.Errorf("limit is 1 to %d", MostResults)
|
||||
}
|
||||
accounts, grants, err := b.settings()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
caller := b.caller(args)
|
||||
granted := grants.Granted(caller, accounts)
|
||||
if account != "" {
|
||||
if !grants.Allows(caller, account) {
|
||||
return nil, grants.refuse(caller, account)
|
||||
}
|
||||
granted = []string{account}
|
||||
}
|
||||
if len(granted) == 0 {
|
||||
if grants.unread != nil {
|
||||
return nil, grants.refuse(caller, "")
|
||||
}
|
||||
return nil, fmt.Errorf("refused: %s is granted no mail account; the operator grants one through the setting grants", caller)
|
||||
}
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second)
|
||||
defer cancel()
|
||||
st, err := b.Index(ctx)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("the index cannot be reached: %v", err)
|
||||
}
|
||||
found, err := st.Search(ctx, Query{Text: query, Accounts: granted, Folder: folder, Limit: limit})
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
out := make([]map[string]any, 0, len(found))
|
||||
for _, m := range found {
|
||||
out = append(out, metadata(m))
|
||||
}
|
||||
return map[string]any{"searched": granted, "found": out}, nil
|
||||
}
|
||||
|
||||
func (b *Bundle) readTool(args map[string]any) (any, error) {
|
||||
if err := only(args, "account", "folder", "uid", "attachment"); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
account, _ := args["account"].(string)
|
||||
folder, _ := args["folder"].(string)
|
||||
attachment, _ := args["attachment"].(string)
|
||||
uid, err := number(args, "uid", 0)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if account == "" || folder == "" || uid <= 0 {
|
||||
return nil, fmt.Errorf("%s needs an account, a folder and a uid", ReadTool)
|
||||
}
|
||||
_, grants, err := b.settings()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
caller := b.caller(args)
|
||||
if !grants.Allows(caller, account) {
|
||||
return nil, grants.refuse(caller, account)
|
||||
}
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
|
||||
defer cancel()
|
||||
st, err := b.Index(ctx)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("the index cannot be reached: %v", err)
|
||||
}
|
||||
m, ok, err := st.Get(ctx, account, folder, uint32(uid))
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if !ok {
|
||||
return nil, fmt.Errorf("the account %s holds no message %d in %s", account, uid, folder)
|
||||
}
|
||||
if attachment != "" {
|
||||
var ref *AttachmentRef
|
||||
for i := range m.Attachments {
|
||||
if m.Attachments[i].Key == attachment {
|
||||
ref = &m.Attachments[i]
|
||||
}
|
||||
}
|
||||
if ref == nil {
|
||||
return nil, fmt.Errorf("the message names no attachment %s", attachment)
|
||||
}
|
||||
if ref.Size > MostAttachment {
|
||||
return nil, fmt.Errorf("the attachment %s is %d bytes, more than the %d this tool answers", ref.Name, ref.Size, MostAttachment)
|
||||
}
|
||||
o, err := b.Objects()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
data, err := o.Get(ctx, ref.Key)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return map[string]any{"attachment": ref, "base64": base64.StdEncoding.EncodeToString(data)}, nil
|
||||
}
|
||||
out := metadata(m)
|
||||
body := m.Body
|
||||
if len(body) > MostBodyAnswered {
|
||||
body = strings.ToValidUTF8(body[:MostBodyAnswered], "") + "\n[cut at 64 KiB]"
|
||||
}
|
||||
out["text"] = body
|
||||
return out, nil
|
||||
}
|
||||
|
||||
func refused(caller, account string) error {
|
||||
return fmt.Errorf("refused: %s is not granted the mail account %s; the operator grants it through the setting grants", caller, account)
|
||||
}
|
||||
|
||||
// ---- arguments ----------------------------------------------------------------------------------------------------
|
||||
|
||||
// only refuses an argument the tool does not take: never ignored, and none can say who calls.
|
||||
func only(args map[string]any, takes ...string) error {
|
||||
for k := range args {
|
||||
if !contains(takes, k) {
|
||||
return fmt.Errorf("the argument %q is not taken", k)
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func number(args map[string]any, key string, otherwise int) (int, error) {
|
||||
v, ok := args[key]
|
||||
if !ok || v == nil {
|
||||
return otherwise, nil
|
||||
}
|
||||
f, ok := v.(float64)
|
||||
if !ok || f != float64(int(f)) {
|
||||
return 0, errors.New(key + " is a whole number")
|
||||
}
|
||||
return int(f), nil
|
||||
}
|
||||
|
||||
func text(d string) map[string]any { return map[string]any{"type": "string", "description": d} }
|
||||
func integer(d string) map[string]any { return map[string]any{"type": "integer", "description": d} }
|
||||
|
||||
func schema(required []string, props map[string]any) map[string]any {
|
||||
s := map[string]any{"type": "object", "properties": props, "additionalProperties": false}
|
||||
if len(required) > 0 {
|
||||
s["required"] = required
|
||||
}
|
||||
return s
|
||||
}
|
||||
@@ -0,0 +1,275 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"bufio"
|
||||
"context"
|
||||
"encoding/base64"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"io"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
stdio "git.novox.be/novox/mesh-sdk/go"
|
||||
)
|
||||
|
||||
func writeFile(t *testing.T, path, content string) {
|
||||
t.Helper()
|
||||
if err := os.WriteFile(path, []byte(content), 0o600); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
|
||||
// runtime plays the node's runtime against the bundle over its MCP channel, as the bundle is launched: it calls tools
|
||||
// naming who called in `_meta`, as the runtime does from the subject a call arrived on, and takes what the bundle
|
||||
// publishes.
|
||||
type runtime struct {
|
||||
t *testing.T
|
||||
in *io.PipeWriter
|
||||
mu sync.Mutex
|
||||
next int
|
||||
answers map[string]chan map[string]any
|
||||
published []map[string]any
|
||||
refuse bool
|
||||
}
|
||||
|
||||
func startBundle(t *testing.T, tools []stdio.Tool) *runtime {
|
||||
t.Helper()
|
||||
inR, inW := io.Pipe()
|
||||
outR, outW := io.Pipe()
|
||||
r := &runtime{t: t, in: inW, answers: map[string]chan map[string]any{}}
|
||||
go func() { _ = stdio.ServeOn(Module, tools, inR, outW) }()
|
||||
go func() {
|
||||
sc := bufio.NewScanner(outR)
|
||||
sc.Buffer(make([]byte, 1<<20), 1<<24)
|
||||
for sc.Scan() {
|
||||
var m map[string]any
|
||||
if json.Unmarshal(sc.Bytes(), &m) != nil {
|
||||
continue
|
||||
}
|
||||
id, _ := json.Marshal(m["id"])
|
||||
if m["method"] == "mesh/publish" {
|
||||
r.mu.Lock()
|
||||
refuse := r.refuse
|
||||
if !refuse {
|
||||
r.published = append(r.published, m["params"].(map[string]any))
|
||||
}
|
||||
r.mu.Unlock()
|
||||
reply := map[string]any{"id": m["id"], "result": map[string]any{}}
|
||||
if refuse {
|
||||
reply = map[string]any{"id": m["id"], "error": map[string]any{"code": -32000, "message": "the bus is away"}}
|
||||
}
|
||||
r.send(reply)
|
||||
continue
|
||||
}
|
||||
r.mu.Lock()
|
||||
ch := r.answers[string(id)]
|
||||
r.mu.Unlock()
|
||||
if ch != nil {
|
||||
ch <- m
|
||||
}
|
||||
}
|
||||
}()
|
||||
t.Cleanup(func() { _ = inW.Close(); _ = outW.Close() })
|
||||
// Answered once the bundle serves: from then on, what it emits goes through this runtime and no other.
|
||||
r.mu.Lock()
|
||||
ready := make(chan map[string]any, 1)
|
||||
r.answers[`"init"`] = ready
|
||||
r.mu.Unlock()
|
||||
r.send(map[string]any{"jsonrpc": "2.0", "id": "init", "method": "initialize", "params": map[string]any{}})
|
||||
select {
|
||||
case <-ready:
|
||||
case <-time.After(10 * time.Second):
|
||||
t.Fatal("the bundle did not start")
|
||||
}
|
||||
return r
|
||||
}
|
||||
|
||||
func (r *runtime) send(m map[string]any) {
|
||||
raw, _ := json.Marshal(m)
|
||||
r.mu.Lock()
|
||||
defer r.mu.Unlock()
|
||||
_, _ = r.in.Write(append(raw, '\n'))
|
||||
}
|
||||
|
||||
// call calls a tool as the caller, answering its text and whether it was an error.
|
||||
func (r *runtime) call(tool, caller string, args map[string]any) (string, bool) {
|
||||
r.mu.Lock()
|
||||
r.next++
|
||||
id := r.next
|
||||
ch := make(chan map[string]any, 1)
|
||||
r.answers[fmt.Sprint(id)] = ch
|
||||
r.mu.Unlock()
|
||||
r.send(map[string]any{"jsonrpc": "2.0", "id": id, "method": "tools/call",
|
||||
"params": map[string]any{"name": tool, "arguments": args, "_meta": map[string]any{stdio.MetaCaller: caller}}})
|
||||
select {
|
||||
case m := <-ch:
|
||||
res := m["result"].(map[string]any)
|
||||
text := res["content"].([]any)[0].(map[string]any)["text"].(string)
|
||||
isErr, _ := res["isError"].(bool)
|
||||
return text, isErr
|
||||
case <-time.After(10 * time.Second):
|
||||
r.t.Fatalf("%s did not answer", tool)
|
||||
}
|
||||
return "", true
|
||||
}
|
||||
|
||||
// aBundle is the bundle over an index holding one message with an attachment, granted to the module triage alone.
|
||||
func aBundle(t *testing.T) (*Bundle, *memStore, *memObjects) {
|
||||
t.Helper()
|
||||
dir := t.TempDir()
|
||||
env := Env{Accounts: filepath.Join(dir, "accounts.json"), Grants: filepath.Join(dir, "grants.json"), Secrets: dir,
|
||||
Status: filepath.Join(dir, "health.json")}
|
||||
writeFile(t, env.Accounts, stageOne)
|
||||
writeFile(t, env.Grants, `{"grants": {"mesh": ["triage"]}}`)
|
||||
st, objects := newMemStore(), newMemObjects()
|
||||
ctx := context.Background()
|
||||
ref, err := Keep(ctx, objects, []byte(attachmentData), "application/pdf", "invoice.pdf")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
m := Message{Account: "mesh", Folder: "INBOX", UID: 7, UIDValidity: 1, MessageID: "att@example.org", Subject: "the invoice",
|
||||
Date: judgedAt, From: []Person{{Name: "Bob", Address: "bob@example.org"}}, To: []Person{{Address: "jschoubben@novox.be"}},
|
||||
Flags: []string{}, Attachments: []AttachmentRef{ref}, Body: "Please find the invoice attached."}
|
||||
if err := st.Add(ctx, m, addedEvent(m)); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
_ = st.SetState(ctx, AccountState{Account: "mesh", State: AccountCurrent, Why: "watching", Heard: judgedAt, Said: judgedAt})
|
||||
b := &Bundle{Env: env, Index: func(context.Context) (Store, error) { return st, nil },
|
||||
Objects: func() (Objects, error) { return objects, nil }, Now: func() time.Time { return judgedAt }}
|
||||
return b, st, objects
|
||||
}
|
||||
|
||||
// The body tools answer a granted caller and refuse every other (ADR 0359 §4, "How it is checked"): the caller is the
|
||||
// runtime's word, never an argument.
|
||||
func TestTheBodyToolsAnswerAGrantedCallerAndRefuseAnyOther(t *testing.T) {
|
||||
b, _, _ := aBundle(t)
|
||||
rt := startBundle(t, b.Tools())
|
||||
read := map[string]any{"account": "mesh", "folder": "INBOX", "uid": 7}
|
||||
|
||||
text, isErr := rt.call(ReadTool, "novox.triage", read)
|
||||
if isErr || !strings.Contains(text, "Please find the invoice attached.") || !strings.Contains(text, `"subject":"the invoice"`) {
|
||||
t.Fatalf("granted: %v %s", isErr, text)
|
||||
}
|
||||
for _, caller := range []string{"novox.other", "person.someone", Unproven} {
|
||||
text, isErr := rt.call(ReadTool, caller, read)
|
||||
if !isErr || !strings.Contains(text, "refused") || strings.Contains(text, "Please find") {
|
||||
t.Fatalf("%s: %v %s", caller, isErr, text)
|
||||
}
|
||||
text, isErr = rt.call(SearchTool, caller, map[string]any{"query": "invoice"})
|
||||
if !isErr || !strings.Contains(text, "refused") {
|
||||
t.Fatalf("%s searched: %v %s", caller, isErr, text)
|
||||
}
|
||||
}
|
||||
// An argument cannot name the caller.
|
||||
if text, isErr := rt.call(ReadTool, "novox.other", map[string]any{"account": "mesh", "folder": "INBOX", "uid": 7, "caller": "novox.triage"}); !isErr || strings.Contains(text, "Please find") {
|
||||
t.Fatalf("an argument named the caller: %s", text)
|
||||
}
|
||||
|
||||
text, isErr = rt.call(SearchTool, "shanks.triage", map[string]any{"query": "invoice"})
|
||||
if isErr || !strings.Contains(text, `"message-id":"att@example.org"`) || strings.Contains(text, "Please find") {
|
||||
t.Fatalf("search: %v %s", isErr, text)
|
||||
}
|
||||
|
||||
sum := ""
|
||||
if m, _ := b.Index(context.Background()); m != nil {
|
||||
got, _, _ := m.Get(context.Background(), "mesh", "INBOX", 7)
|
||||
sum = got.Attachments[0].Key
|
||||
}
|
||||
text, isErr = rt.call(ReadTool, "novox.triage", map[string]any{"account": "mesh", "folder": "INBOX", "uid": 7, "attachment": sum})
|
||||
var att struct {
|
||||
Base64 string `json:"base64"`
|
||||
}
|
||||
if isErr || json.Unmarshal([]byte(text), &att) != nil {
|
||||
t.Fatalf("attachment: %v %s", isErr, text)
|
||||
}
|
||||
if data, _ := base64.StdEncoding.DecodeString(att.Base64); string(data) != attachmentData {
|
||||
t.Fatalf("the attachment read is %q", data)
|
||||
}
|
||||
}
|
||||
|
||||
// mail_accounts says each account's state to anyone, and its address only to a caller granted it.
|
||||
func TestTheAccountsToolListsOnlyTheAccountsTheCallerIsGranted(t *testing.T) {
|
||||
b, _, _ := aBundle(t)
|
||||
rt := startBundle(t, b.Tools())
|
||||
_ = b.Env // the index holds a folder of the account, so an ungranted answer has something to leak
|
||||
st, _ := b.Index(context.Background())
|
||||
_ = st.SetFolder(context.Background(), Folder{Account: "mesh", Name: "Secret-Project", Messages: 41, Synced: judgedAt})
|
||||
_ = st.SetState(context.Background(), AccountState{Account: "mesh", State: AccountFailing, Why: "the server said NO [AUTHENTICATIONFAILED]", Said: judgedAt})
|
||||
text, isErr := rt.call(AccountsTool, "novox.other", map[string]any{})
|
||||
if isErr || !strings.Contains(text, `"accounts":[]`) {
|
||||
t.Fatalf("ungranted: %s", text)
|
||||
}
|
||||
for _, leak := range []string{"mesh\"", "jschoubben@", "Secret-Project", "41", "AUTHENTICATIONFAILED", "failing"} {
|
||||
if strings.Contains(text, leak) {
|
||||
t.Fatalf("an ungranted caller learns %q: %s", leak, text)
|
||||
}
|
||||
}
|
||||
text, _ = rt.call(AccountsTool, "novox.triage", map[string]any{})
|
||||
if !strings.Contains(text, `"address":"jschoubben@novox.be"`) || !strings.Contains(text, "Secret-Project") || !strings.Contains(text, "AUTHENTICATIONFAILED") {
|
||||
t.Fatalf("granted: %s", text)
|
||||
}
|
||||
}
|
||||
|
||||
// The events the keeper wrote are said on the bus as the module, through the runtime, with their metadata and never a
|
||||
// body; each is removed once the bus took it, and kept when it did not.
|
||||
func TestTheEventsAreSaidOnTheBusThroughTheRuntime(t *testing.T) {
|
||||
b, st, _ := aBundle(t)
|
||||
rt := startBundle(t, b.Tools())
|
||||
p := &Publisher{Index: b.Index, Emit: func(key string, body any) error { return stdio.Emit(key, body) }, Logf: t.Logf}
|
||||
ctx := context.Background()
|
||||
|
||||
rt.mu.Lock()
|
||||
rt.refuse = true
|
||||
rt.mu.Unlock()
|
||||
if n, err := p.Once(ctx); n != 0 || err == nil {
|
||||
t.Fatalf("said %d with the bus away (%v)", n, err)
|
||||
}
|
||||
if left, _ := st.Pending(ctx, 10); len(left) != 1 {
|
||||
t.Fatalf("an event the bus did not take was dropped: %d left", len(left))
|
||||
}
|
||||
|
||||
rt.mu.Lock()
|
||||
rt.refuse = false
|
||||
rt.mu.Unlock()
|
||||
if n, err := p.Once(ctx); n != 1 || err != nil {
|
||||
t.Fatalf("said %d (%v)", n, err)
|
||||
}
|
||||
if left, _ := st.Pending(ctx, 10); len(left) != 0 {
|
||||
t.Fatalf("%d event(s) left after the bus took them", len(left))
|
||||
}
|
||||
rt.mu.Lock()
|
||||
defer rt.mu.Unlock()
|
||||
if len(rt.published) != 1 {
|
||||
t.Fatalf("published %d", len(rt.published))
|
||||
}
|
||||
pub := rt.published[0]
|
||||
raw, _ := json.Marshal(pub)
|
||||
if pub["key"] != "mail.mesh.message.added" || !strings.Contains(string(raw), `"subject":"the invoice"`) ||
|
||||
strings.Contains(string(raw), "Please find") || strings.Contains(string(raw), "dGhlIGludm9pY2Ug") {
|
||||
t.Fatalf("published %s", raw)
|
||||
}
|
||||
}
|
||||
|
||||
// A grants setting that cannot be read grants nobody, and every refusal says why (issue 733).
|
||||
func TestARefusalSaysWhyTheGrantsCouldNotBeRead(t *testing.T) {
|
||||
b, _, _ := aBundle(t)
|
||||
writeFile(t, b.Env.Grants, `{"grants": {"mesh": "triage"}}`)
|
||||
rt := startBundle(t, b.Tools())
|
||||
text, isErr := rt.call(ReadTool, "novox.triage", map[string]any{"account": "mesh", "folder": "INBOX", "uid": 7})
|
||||
if !isErr || !strings.Contains(text, "the setting grants cannot be read") || !strings.Contains(text, "not the module's") || strings.Contains(text, "Please find") {
|
||||
t.Fatalf("read: %v %s", isErr, text)
|
||||
}
|
||||
text, isErr = rt.call(SearchTool, "novox.triage", map[string]any{"query": "invoice"})
|
||||
if !isErr || !strings.Contains(text, "the setting grants cannot be read") {
|
||||
t.Fatalf("search: %v %s", isErr, text)
|
||||
}
|
||||
text, _ = rt.call(AccountsTool, "novox.triage", map[string]any{})
|
||||
if !strings.Contains(text, "the setting grants cannot be read") || !strings.Contains(text, `"accounts":[]`) {
|
||||
t.Fatalf("accounts: %s", text)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,19 @@
|
||||
module mesh-mail
|
||||
|
||||
go 1.25.0
|
||||
|
||||
require (
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261013182924-5933eb757d44
|
||||
github.com/emersion/go-imap/v2 v2.0.0-beta.8
|
||||
github.com/emersion/go-message v0.18.2
|
||||
github.com/jackc/pgx/v5 v5.11.0
|
||||
)
|
||||
|
||||
require (
|
||||
github.com/emersion/go-sasl v0.0.0-20241020182733-b788ff22d5a6 // indirect
|
||||
github.com/jackc/pgpassfile v1.0.0 // indirect
|
||||
github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 // indirect
|
||||
github.com/jackc/puddle/v2 v2.2.2 // indirect
|
||||
golang.org/x/sync v0.17.0 // indirect
|
||||
golang.org/x/text v0.29.0 // indirect
|
||||
)
|
||||
@@ -0,0 +1,65 @@
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261013182924-5933eb757d44 h1:r5xIo5qWOXQdCSMxvqJy8kV4U+P/QWXFK6971Ha6wU8=
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261013182924-5933eb757d44/go.mod h1:GFuZUElBZ9A++mxgIKo97aXXo+kV0uJ/UkbhQPPIbrY=
|
||||
github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
|
||||
github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
|
||||
github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
|
||||
github.com/emersion/go-imap/v2 v2.0.0-beta.8 h1:5IXZK1E33DyeP526320J3RS7eFlCYGFgtbrfapqDPug=
|
||||
github.com/emersion/go-imap/v2 v2.0.0-beta.8/go.mod h1:dhoFe2Q0PwLrMD7oZw8ODuaD0vLYPe5uj2wcOMnvh48=
|
||||
github.com/emersion/go-message v0.18.2 h1:rl55SQdjd9oJcIoQNhubD2Acs1E6IzlZISRTK7x/Lpg=
|
||||
github.com/emersion/go-message v0.18.2/go.mod h1:XpJyL70LwRvq2a8rVbHXikPgKj8+aI0kGdHlg16ibYA=
|
||||
github.com/emersion/go-sasl v0.0.0-20241020182733-b788ff22d5a6 h1:oP4q0fw+fOSWn3DfFi4EXdT+B+gTtzx8GC9xsc26Znk=
|
||||
github.com/emersion/go-sasl v0.0.0-20241020182733-b788ff22d5a6/go.mod h1:iL2twTeMvZnrg54ZoPDNfJaJaqy0xIQFuBdrLsmspwQ=
|
||||
github.com/jackc/pgpassfile v1.0.0 h1:/6Hmqy13Ss2zCq62VdNG8tM1wchn8zjSGOBJ6icpsIM=
|
||||
github.com/jackc/pgpassfile v1.0.0/go.mod h1:CEx0iS5ambNFdcRtxPj5JhEz+xB6uRky5eyVu/W2HEg=
|
||||
github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 h1:iCEnooe7UlwOQYpKFhBabPMi4aNAfoODPEFNiAnClxo=
|
||||
github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761/go.mod h1:5TJZWKEWniPve33vlWYSoGYefn3gLQRzjfDlhSJ9ZKM=
|
||||
github.com/jackc/pgx/v5 v5.11.0 h1:IzBBtyK9AHqf98cctWFifYSci2hgQR/cd56wB4p+ogg=
|
||||
github.com/jackc/pgx/v5 v5.11.0/go.mod h1:mal1tBGAFfLHvZzaYh77YS/eC6IX9OWbRV1QIIM0Jn4=
|
||||
github.com/jackc/puddle/v2 v2.2.2 h1:PR8nw+E/1w0GLuRFSmiioY6UooMp6KJv0/61nB7icHo=
|
||||
github.com/jackc/puddle/v2 v2.2.2/go.mod h1:vriiEXHvEE654aYKXXjOvZM39qJ0q+azkZFrfEOc3H4=
|
||||
github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
|
||||
github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
|
||||
github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
|
||||
github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI=
|
||||
github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
|
||||
github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U=
|
||||
github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U=
|
||||
github.com/yuin/goldmark v1.4.13/go.mod h1:6yULJ656Px+3vBD8DxQVa3kxgyrAnzto9xy5taEt/CY=
|
||||
golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
|
||||
golang.org/x/crypto v0.0.0-20210921155107-089bfa567519/go.mod h1:GvvjBRRGRdwPK5ydBHafDWAxML/pGHZbMvKqRZ5+Abc=
|
||||
golang.org/x/mod v0.6.0-dev.0.20220419223038-86c51ed26bb4/go.mod h1:jJ57K6gSWd91VN4djpZkiMVwK6gcyfeH4XE8wZrZaV4=
|
||||
golang.org/x/mod v0.8.0/go.mod h1:iBbtSCu2XBx23ZKBPSOrRkjjQPZFPuis4dIYUhu/chs=
|
||||
golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
|
||||
golang.org/x/net v0.0.0-20210226172049-e18ecbb05110/go.mod h1:m0MpNAwzfU5UDzcl9v0D8zg8gWTRqZa9RBIspLL5mdg=
|
||||
golang.org/x/net v0.0.0-20220722155237-a158d28d115b/go.mod h1:XRhObCWvk6IyKnWLug+ECip1KBveYUHfp+8e9klMJ9c=
|
||||
golang.org/x/net v0.6.0/go.mod h1:2Tu9+aMcznHK/AK1HMvgo6xiTLG5rD5rZLDS+rp2Bjs=
|
||||
golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
|
||||
golang.org/x/sync v0.0.0-20220722155255-886fb9371eb4/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
|
||||
golang.org/x/sync v0.1.0/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
|
||||
golang.org/x/sync v0.17.0 h1:l60nONMj9l5drqw6jlhIELNv9I0A4OFgRsG9k2oT9Ug=
|
||||
golang.org/x/sync v0.17.0/go.mod h1:9KTHXmSnoGruLpwFjVSX0lNNA75CykiMECbovNTZqGI=
|
||||
golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
|
||||
golang.org/x/sys v0.0.0-20201119102817-f84b799fce68/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
|
||||
golang.org/x/sys v0.0.0-20210615035016-665e8c7367d1/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
|
||||
golang.org/x/sys v0.0.0-20220520151302-bc2c85ada10a/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
|
||||
golang.org/x/sys v0.0.0-20220722155257-8c9f86f7a55f/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
|
||||
golang.org/x/sys v0.5.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
|
||||
golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
|
||||
golang.org/x/term v0.0.0-20210927222741-03fcf44c2211/go.mod h1:jbD1KX2456YbFQfuXm/mYQcufACuNUgVhRMnK/tPxf8=
|
||||
golang.org/x/term v0.5.0/go.mod h1:jMB1sMXY+tzblOD4FWmEbocvup2/aLOaQEp7JmGp78k=
|
||||
golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
|
||||
golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
|
||||
golang.org/x/text v0.3.7/go.mod h1:u+2+/6zg+i71rQMx5EYifcz6MCKuco9NR6JIITiCfzQ=
|
||||
golang.org/x/text v0.7.0/go.mod h1:mrYo+phRRbMaCq/xk9113O4dZlRixOauAjOtrjsXDZ8=
|
||||
golang.org/x/text v0.14.0/go.mod h1:18ZOQIKpY8NJVqYksKHtTdi31H5itFRjB5/qKTNYzSU=
|
||||
golang.org/x/text v0.29.0 h1:1neNs90w9YzJ9BocxfsQNHKuAT4pkghyXc4nhZ6sJvk=
|
||||
golang.org/x/text v0.29.0/go.mod h1:7MhJOA9CD2qZyOKYazxdYMF85OwPdEr9jTtBpO7ydH4=
|
||||
golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
|
||||
golang.org/x/tools v0.0.0-20191119224855-298f0cb1881e/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
|
||||
golang.org/x/tools v0.1.12/go.mod h1:hNGJHUnrk76NpqgfD5Aqm5Crs+Hm0VOH/i9J2+nxYbc=
|
||||
golang.org/x/tools v0.6.0/go.mod h1:Xwgl3UAJ/d3gWutnCtw505GrjyAbvKui8lOU390QaIU=
|
||||
golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
|
||||
gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
|
||||
gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
|
||||
gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
|
||||
gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
|
||||
@@ -0,0 +1,189 @@
|
||||
{
|
||||
"module": "mesh-mail",
|
||||
"version": "1",
|
||||
"slug": "mail",
|
||||
"requires": [
|
||||
"postgres-database",
|
||||
"s3-bucket"
|
||||
],
|
||||
"contributes": {
|
||||
"postgres-database": {
|
||||
"name": "mail"
|
||||
}
|
||||
},
|
||||
"binds": {
|
||||
"postgres-database": "${dir:state}/database.json",
|
||||
"s3-bucket": "${dir:state}/store-binding.json"
|
||||
},
|
||||
"secrets": {
|
||||
"postgres-database": "${dir:state}/database.secret",
|
||||
"s3-bucket": "${dir:state}/store.secret"
|
||||
},
|
||||
"own-secrets": {
|
||||
"mail-credential-*": {
|
||||
"path": "${dir:state}/mail-credential-*.secret",
|
||||
"issued-by": "outside"
|
||||
}
|
||||
},
|
||||
"tools": [
|
||||
"mail_accounts",
|
||||
"mail_search",
|
||||
"mail_read",
|
||||
"mail_health"
|
||||
],
|
||||
"streams": [
|
||||
{
|
||||
"name": "mail",
|
||||
"events": [
|
||||
"mail.*.message.added",
|
||||
"mail.*.message.changed",
|
||||
"mail.*.message.removed"
|
||||
],
|
||||
"max-age-days": 7,
|
||||
"grant": {
|
||||
"config": "grants",
|
||||
"events": "mail.{key}.**"
|
||||
}
|
||||
}
|
||||
],
|
||||
"resources": [
|
||||
{
|
||||
"id": "state",
|
||||
"type": "directory",
|
||||
"mode": "0700",
|
||||
"place": "."
|
||||
},
|
||||
{
|
||||
"id": "database-url",
|
||||
"type": "file",
|
||||
"path": "${dir:state}/database.url",
|
||||
"mode": "0600",
|
||||
"content": "postgresql://${bound:postgres-database:as}:${secret:postgres-database}@${bound:postgres-database:at}:${bound:postgres-database:port}/${bound:postgres-database:as}\n"
|
||||
},
|
||||
{
|
||||
"id": "store",
|
||||
"type": "file",
|
||||
"path": "${dir:state}/store.json",
|
||||
"mode": "0600",
|
||||
"content": "{\n \"scheme\": \"${bound:s3-bucket:scheme}\",\n \"at\": \"${bound:s3-bucket:at}\",\n \"port\": \"${bound:s3-bucket:port}\",\n \"bucket\": \"${bound:s3-bucket:bucket}\",\n \"region\": \"${bound:s3-bucket:region}\",\n \"key\": \"${bound:s3-bucket:as}\",\n \"secret\": \"${secret:s3-bucket}\"\n}\n"
|
||||
},
|
||||
{
|
||||
"id": "accounts",
|
||||
"trusted": true,
|
||||
"type": "file",
|
||||
"path": "${dir:state}/accounts.json",
|
||||
"mode": "0600",
|
||||
"merge": "json",
|
||||
"takes-settings": [
|
||||
"accounts"
|
||||
],
|
||||
"content": "{\n \"accounts\": {}\n}\n"
|
||||
},
|
||||
{
|
||||
"id": "grants",
|
||||
"trusted": true,
|
||||
"type": "file",
|
||||
"path": "${dir:state}/grants.json",
|
||||
"mode": "0600",
|
||||
"merge": "json",
|
||||
"takes-settings": [
|
||||
"grants"
|
||||
],
|
||||
"content": "{\n \"grants\": {}\n}\n"
|
||||
},
|
||||
{
|
||||
"id": "prepare",
|
||||
"type": "process",
|
||||
"name": "mesh-mail-prepare",
|
||||
"artifact": "tools",
|
||||
"run": [
|
||||
"./mesh-mail",
|
||||
"prepare"
|
||||
],
|
||||
"run-once": true,
|
||||
"env": {
|
||||
"DATABASE_URL_FILE": "${dir:state}/database.url"
|
||||
},
|
||||
"restart-on": [
|
||||
"database-url"
|
||||
]
|
||||
},
|
||||
{
|
||||
"id": "keep",
|
||||
"type": "process",
|
||||
"name": "mesh-mail-keep",
|
||||
"artifact": "tools",
|
||||
"run": [
|
||||
"./mesh-mail",
|
||||
"keep"
|
||||
],
|
||||
"env": {
|
||||
"DATABASE_URL_FILE": "${dir:state}/database.url",
|
||||
"MESH_MAIL_ACCOUNTS": "${dir:state}/accounts.json",
|
||||
"MESH_MAIL_SECRETS": "${dir:state}",
|
||||
"MESH_MAIL_STORE": "${dir:state}/store.json"
|
||||
},
|
||||
"restart-on": [
|
||||
"database-url",
|
||||
"store",
|
||||
"accounts"
|
||||
],
|
||||
"health": {
|
||||
"kind": "unit"
|
||||
}
|
||||
},
|
||||
{
|
||||
"id": "watch",
|
||||
"type": "process",
|
||||
"name": "mesh-mail-watch",
|
||||
"artifact": "tools",
|
||||
"run": [
|
||||
"./mesh-mail",
|
||||
"watch"
|
||||
],
|
||||
"user": "${machine:account}",
|
||||
"env": {
|
||||
"DATABASE_URL_FILE": "${dir:state}/database.url",
|
||||
"MESH_MAIL_ACCOUNTS": "${dir:state}/accounts.json",
|
||||
"MESH_MAIL_SECRETS": "${dir:state}",
|
||||
"MESH_MAIL_STATUS": "${dir:state}/health.json"
|
||||
},
|
||||
"restart-on": [
|
||||
"database-url",
|
||||
"accounts"
|
||||
],
|
||||
"health": {
|
||||
"kind": "tool",
|
||||
"tool": "mail_health",
|
||||
"needs": "postgres-database",
|
||||
"interval": "60s",
|
||||
"timeout": "10s",
|
||||
"looks": 2,
|
||||
"grace": "180s"
|
||||
}
|
||||
}
|
||||
],
|
||||
"build": {
|
||||
"artifacts": [
|
||||
{
|
||||
"name": "tools",
|
||||
"kind": "bundle",
|
||||
"language": "go",
|
||||
"system": "arch",
|
||||
"from": "cmd/mesh-mail",
|
||||
"binary": "mesh-mail",
|
||||
"loads": [
|
||||
"mesh-mail"
|
||||
],
|
||||
"env": {
|
||||
"DATABASE_URL_FILE": "${dir:state}/database.url",
|
||||
"MESH_MAIL_ACCOUNTS": "${dir:state}/accounts.json",
|
||||
"MESH_MAIL_GRANTS": "${dir:state}/grants.json",
|
||||
"MESH_MAIL_SECRETS": "${dir:state}",
|
||||
"MESH_MAIL_STORE": "${dir:state}/store.json",
|
||||
"MESH_MAIL_STATUS": "${dir:state}/health.json"
|
||||
}
|
||||
}
|
||||
]
|
||||
}
|
||||
}
|
||||
@@ -20,7 +20,7 @@ WORKDIR /src
|
||||
COPY go.mod go.sum ./
|
||||
RUN go mod download
|
||||
COPY . .
|
||||
RUN CGO_ENABLED=0 GOOS=linux go build -trimpath -buildvcs=false -ldflags=-buildid= -o /mesh-route-proxy ./examples/route-proxy
|
||||
RUN CGO_ENABLED=0 GOOS=linux go build -trimpath -o /mesh-route-proxy ./examples/route-proxy
|
||||
|
||||
# A small runtime with the public CA roots the ACME client needs to reach a real authority, and run
|
||||
# as root so it can bind :80 and :443 — the two privileged ports a public front door listens on.
|
||||
|
||||
Reference in New Issue
Block a user