Mouth: basis "follow" for engagement peers (CHAT_FOLLOW, default off) #74
4 changed files with 512 additions and 8 deletions
|
|
@ -206,7 +206,7 @@ A client asks for the facts by the kind it polls with: `gamecult.persona.chat_pe
|
|||
|
||||
| Field | Type | Meaning |
|
||||
|---|---|---|
|
||||
| 12 `basis` | `"direct"`, `"tag"`, `"reply"`, `"room"` or nil | Why the message is for this Persona, by what Matrix says, first that holds: `direct` the room is one of her DMs; `tag` `m.mentions.user_ids` lists her id, or the client sent no `m.mentions` at all and an `<a href>` in the HTML body is exactly her `matrix.to` link; `reply` it replies to, or is in a thread under, an event her account wrote (her Self run speaks as her account, so its words count; any member's message in a thread under one of her events is a `reply` too, and the target is read once per room and event and kept for a minute, hit or failure); `room` an `@room` mention. A `notice` or an `edit` costs no reply lookup. Known limit: lookups run one at a time inside the `/sync` round (10 s each, at most two per reply), so a slow homeserver slows the round; they are not parallelised or capped per round. |
|
||||
| 12 `basis` | `"direct"`, `"tag"`, `"reply"`, `"room"`, `"follow"` (only with `CHAT_FOLLOW=1`, see Follow) or nil | Why the message is for this Persona, by what Matrix says, first that holds: `direct` the room is one of her DMs; `tag` `m.mentions.user_ids` lists her id, or the client sent no `m.mentions` at all and an `<a href>` in the HTML body is exactly her `matrix.to` link; `reply` it replies to, or is in a thread under, an event her account wrote (her Self run speaks as her account, so its words count; any member's message in a thread under one of her events is a `reply` too, and the target is read once per room and event and kept for a minute, hit or failure); `room` an `@room` mention. A `notice` or an `edit` costs no reply lookup. Known limit: lookups run one at a time inside the `/sync` round (10 s each, at most two per reply), so a slow homeserver slows the round; they are not parallelised or capped per round. |
|
||||
| 13 `form` | `"message"`, `"notice"` or `"edit"` | `m.notice` is an automated line; `m.replace` is an edit of an earlier message. Only a `message` can address anyone. |
|
||||
|
||||
What the mouth cannot know, and so what a nil `basis` means: **unproven, not "not for her"**. Her handle and public name in the text (any casing, with or without `@`, `'s`), the names her stewardships list, a project's repo name, and whether the sender is the operator or a Mind are all matched by Huginn, not here. A reply whose target the homeserver will not give is nil and logged. The mouth also cannot see whether a message was meant for her without naming her. Every delivery is still delivered, claimed tags included: nothing is dropped in the mouth, so the Huginn rule cannot lose a message the mouth had no way to judge.
|
||||
|
|
@ -253,6 +253,19 @@ A line a delivery already shows and a room she cannot speak in change nothing he
|
|||
|
||||
Golden bytes: `tests/fixtures/persona-chat-v5/` (`chat_pending.v3.tail` carries two lines and no gap, `chat_pending.v3.tail-gaps` one line and two gap rooms; the earlier v3 samples gain an empty fourth element); Huginn pins the same bytes.
|
||||
|
||||
### Follow (`basis` `"follow"`, `CHAT_FOLLOW=1`)
|
||||
|
||||
A message from someone she is in conversation with is for her even when it names no one. **It is off unless the mouth runs with `CHAT_FOLLOW=1`** (exactly `1`; default off), because Huginn's parser refuses an unknown basis until its `Basis::Follow` (and the one `is_for` line) lands. With the flag off the mouth emits no `follow`, makes no extra read, and keeps no follow state.
|
||||
|
||||
With it on:
|
||||
|
||||
- **Anchor.** Her own post in a public room anchors that room for W = 600 s from the post. The mouth anchors at send time (the send's own time, in `send`, in the background: the send receipt never waits on the room-state read) and also from the /sync echo of her own events, so a post from another client of her account anchors too; one event anchors once. Only her posting again re-anchors the room and re-snapshots who is counted; a follow never extends the window.
|
||||
- **Counted senders** (snapshotted at the post): those who spoke in that room in the 10 minutes before the post (a per-room ring of recent senders with event times, 200 entries), plus anyone the post tagged (`m.mentions.user_ids`) or replied to (the author of the quoted `m.in_reply_to`, not a thread fallback pointer) when the event carries it. Someone who first speaks after her post is not counted. A member who spoke in the 10 minutes before her post is counted, bounded by 3 follows per hour per (peer, room). Nothing reads message text; no ids are scraped from bodies. She herself, and bridged senders (the `platformEvent` identifier rule), never count.
|
||||
- **Follow.** A `message` (not an edit, not a notice) from a counted sender gets `follow` when it would otherwise have no basis or `room`; `tag`, `reply` and `direct` are never replaced. Its timestamp must lie inside the window (a homeserver clock up to 60 s behind still follows) and this process's clock must not be past the window's end by more than 60 s, so an old or forged `origin_server_ts` reuses nothing. At most K = 3 per (Mind, peer, room) in the hour before the message; past it the message keeps the basis it had.
|
||||
- **"Public" means non-DM, non-encrypted, non-bridged.** The room is not one of her DMs and its state has no `m.room.encryption`, `m.bridge` or `uk.half-shot.bridge` (read once per room, kept 5 minutes; unreadable or malformed state is not public). An invite-only plaintext room counts as public. Nothing is anchored in, and no follow is delivered in, any other room.
|
||||
|
||||
State is in memory only (per-room anchor, sender ring, follow times), purged on every sync pass; nothing is stored, and a restart only shortens windows.
|
||||
|
||||
### Reply link (`chat_pending.v4`)
|
||||
|
||||
A reply to anyone but the Persona arrives bare in v3: `basis` is `reply` only when the quoted author is her account. v4 adds what the message quotes. **It is off unless the mouth runs with `CHAT_PENDING_V4=1`** (exactly `1`; default off), because Huginn pins the tuple bytes and must read v4 first. With the flag off, `gamecult.persona.chat_pending_request.v3` is not a served kind: it is refused as any unknown kind is (`chat_refusal` `UnknownKind`, logged), byte for byte, and v1/v2/v3 requests are answered as before. With it on, that request kind (the v1 tuple `[persona, after_seq]` again; the kind asked by names the answer) is answered with `gamecult.persona.chat_pending.v4`.
|
||||
|
|
|
|||
378
tests/persona-matrix-mouth-follow.test.mjs
Normal file
378
tests/persona-matrix-mouth-follow.test.mjs
Normal file
|
|
@ -0,0 +1,378 @@
|
|||
// Follow (docs/bridge.md, "Follow"): her own post in a public room anchors it for W; those who spoke there in the W before it (and whoever it tagged or
|
||||
// replied to) are counted, and their later messages are basis "follow", at most K an hour per (peer, room), and only with the mouth's follow flag. The homeserver is an in-process fetch port:
|
||||
// a scripted /sync (one reply per round), the room state and the events a reply target is read from.
|
||||
import test from "node:test";
|
||||
import assert from "node:assert/strict";
|
||||
import { createMouth, followFromEnv } from "../tools/persona-matrix-mouth.mjs";
|
||||
import * as docs from "../tools/persona-chat-documents.mjs";
|
||||
|
||||
const HS = "gamecult.test";
|
||||
const ALIAS = `#swarm:${HS}`;
|
||||
const ROOM = "!swarm:gamecult.test";
|
||||
const OTHER_ALIAS = `#side:${HS}`;
|
||||
const OTHER_ROOM = "!side:gamecult.test";
|
||||
const ME = `@eureka:${HS}`;
|
||||
const KIKO = `@kiko:${HS}`;
|
||||
const STRANGER = `@stranger:${HS}`;
|
||||
const LENA = `@lena:${HS}`;
|
||||
const PLATFORM = ["disc", "ord"].join("");
|
||||
const SEC = 1000, W = 600 * SEC, T0 = 1_000_000_000_000;
|
||||
const msg = (id, sender, ts, content = {}) => ({ type: "m.room.message", event_id: id, sender, origin_server_ts: ts, content: { msgtype: "m.text", body: `line ${id}`, ...content } });
|
||||
const tagKiko = { "m.mentions": { user_ids: [KIKO] } };
|
||||
|
||||
// rounds: sync replies after the baseline, each { [roomId]: events }. state: room id -> state events.
|
||||
function world({ rounds, state = {}, stateStatus = {}, hangState = false, stateDelay = 0, stateFailures = 0, baselineEvents = null, onSync = null, follow = true, rooms = [ALIAS, OTHER_ALIAS], direct } = {}) {
|
||||
const seen = [];
|
||||
const known = new Map();
|
||||
let round = 0;
|
||||
const reply = (status, body) => new Response(JSON.stringify(body), { status, headers: { "content-type": "application/json" } });
|
||||
const hsFetch = async (url, init = {}) => {
|
||||
const u = new URL(url);
|
||||
const p = decodeURIComponent(u.pathname).replace(/^\/_matrix\/client\/v[13]/, "");
|
||||
seen.push(`${init.method ?? "GET"} ${p}`);
|
||||
if (p === "/account/whoami") return reply(200, { user_id: ME });
|
||||
if (p === "/joined_rooms") return reply(200, { joined_rooms: [ROOM, OTHER_ROOM] });
|
||||
if (p.includes("/account_data/m.direct")) return direct ? reply(200, direct) : reply(404, { errcode: "M_NOT_FOUND" });
|
||||
if (p.startsWith("/directory/room/")) return reply(200, { room_id: { [ALIAS]: ROOM, [OTHER_ALIAS]: OTHER_ROOM }[p.slice(16)] });
|
||||
if (p.startsWith("/presence/") || p.includes("/receipt/")) return reply(200, {});
|
||||
const st = p.match(/^\/rooms\/([^/]+)\/state$/);
|
||||
if (st && stateDelay > 0) await new Promise(r => setTimeout(r, stateDelay));
|
||||
if (st && hangState) return new Promise(() => {});
|
||||
if (st && stateFailures > 0) { stateFailures--; return reply(500, { errcode: "M_UNKNOWN" }); }
|
||||
if (st) return stateStatus[st[1]] ? reply(stateStatus[st[1]], { errcode: "M_UNKNOWN" }) : reply(200, state[st[1]] ?? []);
|
||||
if (init.method === "PUT" && p.includes("/send/m.room.message/")) return reply(200, { event_id: "$sent1" });
|
||||
const ev = p.match(/^\/rooms\/([^/]+)\/event\/(.+)$/);
|
||||
if (ev) return known.has(ev[2]) ? reply(200, known.get(ev[2])) : reply(404, { errcode: "M_NOT_FOUND" });
|
||||
if (p === "/sync") {
|
||||
if (u.searchParams.get("since") === null) return reply(200, { next_batch: "b0", ...(baselineEvents ? { rooms: { join: Object.fromEntries(Object.entries(baselineEvents).map(([r, list]) => [r, { timeline: { events: list } }])) } } : {}) });
|
||||
onSync?.(setClock);
|
||||
const events = rounds[round++] ?? {};
|
||||
for (const list of Object.values(events)) for (const e of list) known.set(e.event_id, e);
|
||||
return reply(200, { next_batch: `b${round}`, rooms: { join: Object.fromEntries(Object.entries(events).map(([r, list]) => [r, { timeline: { events: list } }])) } });
|
||||
}
|
||||
if (p.includes("/messages")) return reply(200, { chunk: [] });
|
||||
return reply(404, { errcode: "M_NOT_FOUND" });
|
||||
};
|
||||
let clock = T0;
|
||||
const setClock = t => { clock = t; };
|
||||
const mouth = createMouth({
|
||||
persona: "eureka", homeserver: "http://hs.test", serverName: HS, rooms, token: "tok", fetch: hsFetch, follow,
|
||||
state: { load: async () => null, save: async () => {} }, brake: async () => true, log: () => {}, syncTimeoutMs: 1, now: () => clock,
|
||||
});
|
||||
return { mouth, seen, clock: () => clock, at: t => { clock = t; }, basisOfRound: async () => (await mouth.syncOnce()).map(d => [d.eventId, d.basis]) };
|
||||
}
|
||||
const run = async (w, n) => { await w.mouth.syncOnce(); const out = []; for (let i = 0; i < n; i++) out.push(await w.basisOfRound()); return out; };
|
||||
|
||||
const go = async (w, t) => { w.at(t); return w.basisOfRound(); }; // the process clock moves with the round
|
||||
const mine = (id, ts, content = {}) => msg(id, ME, ts, content);
|
||||
const MIN = 60 * SEC;
|
||||
|
||||
test("follow: a counted sender is one who spoke in the 10 minutes before her post; 9 minutes counts, 11 does not; one who first speaks after does not", async () => {
|
||||
const w = world({ rounds: [
|
||||
{ [ROOM]: [msg("$old", KIKO, T0 - 11 * MIN), msg("$edge", `@edge:${HS}`, T0 - 10.5 * MIN), msg("$recent", LENA, T0 - 9 * MIN)] },
|
||||
{ [ROOM]: [mine("$mine", T0)] },
|
||||
{ [ROOM]: [msg("$k", KIKO, T0 + MIN), msg("$l", LENA, T0 + 2 * MIN), msg("$s", STRANGER, T0 + 3 * MIN), msg("$e", `@edge:${HS}`, T0 + 3 * MIN)] },
|
||||
] });
|
||||
const out = await run(w, 3);
|
||||
assert.deepEqual(out[2].sort(), [["$e", null], ["$k", null], ["$l", "follow"], ["$s", null]]);
|
||||
});
|
||||
|
||||
test("follow: a post's tag and reply target are counted though they never spoke; she herself never is", async () => {
|
||||
const w = world({ rounds: [
|
||||
{ [ROOM]: [msg("$theirs", LENA, T0 - 5 * MIN)] },
|
||||
{ [ROOM]: [mine("$mine", T0, { "m.mentions": { user_ids: [KIKO, ME] }, "m.relates_to": { "m.in_reply_to": { event_id: "$theirs" } } })] },
|
||||
{ [ROOM]: [msg("$k", KIKO, T0 + MIN), msg("$s", STRANGER, T0 + MIN)] },
|
||||
] });
|
||||
assert.deepEqual((await run(w, 3))[2], [["$k", "follow"], ["$s", null]]);
|
||||
});
|
||||
|
||||
|
||||
test("follow: the chain dies when she stops speaking; a follow does not extend W; speaking again re-anchors and re-snapshots", async () => {
|
||||
const w = world({ rounds: [
|
||||
{ [ROOM]: [msg("$k0", KIKO, T0 - MIN)] },
|
||||
{ [ROOM]: [mine("$mine", T0)] },
|
||||
{ [ROOM]: [msg("$a", KIKO, T0 + 9 * MIN)] }, // inside W: follow
|
||||
{ [ROOM]: [msg("$b", KIKO, T0 + 10 * MIN + SEC)] }, // past W from her post though a follow came at 9 min
|
||||
{ [ROOM]: [mine("$again", T0 + 11 * MIN)] },
|
||||
{ [ROOM]: [msg("$c", KIKO, T0 + 20 * MIN), msg("$d", LENA, T0 + 20 * MIN)] },
|
||||
] });
|
||||
await w.mouth.syncOnce();
|
||||
assert.deepEqual(await go(w, T0 - MIN), [["$k0", null]]);
|
||||
assert.deepEqual(await go(w, T0), []);
|
||||
assert.deepEqual(await go(w, T0 + 9 * MIN), [["$a", "follow"]]);
|
||||
assert.deepEqual(await go(w, T0 + 10 * MIN + SEC), [["$b", null]]);
|
||||
assert.deepEqual(await go(w, T0 + 11 * MIN), []);
|
||||
assert.deepEqual(await go(w, T0 + 20 * MIN), [["$c", "follow"], ["$d", null]]); // lena never spoke before the second post either
|
||||
});
|
||||
|
||||
test("follow: the window needs this process's clock too: a message stamped inside it but read long after is not a follow", async () => {
|
||||
const w = world({ rounds: [{ [ROOM]: [msg("$k0", KIKO, T0 - MIN)] }, { [ROOM]: [mine("$mine", T0)] }, { [ROOM]: [msg("$forged", KIKO, T0 + MIN)] }] });
|
||||
await w.mouth.syncOnce();
|
||||
await go(w, T0 - MIN); await go(w, T0);
|
||||
assert.deepEqual(await go(w, T0 + 30 * MIN), [["$forged", null]]);
|
||||
});
|
||||
|
||||
test("follow: her mouth send anchors at send time, and its /sync echo does not anchor again", async () => {
|
||||
const w = world({ rounds: [
|
||||
{ [ROOM]: [msg("$k0", KIKO, T0 - MIN)] },
|
||||
{ [ROOM]: [msg("$sent1", ME, T0 + 8 * MIN, { body: "echo" })] },
|
||||
{ [ROOM]: [msg("$late", KIKO, T0 + 12 * MIN)] },
|
||||
] });
|
||||
await w.mouth.syncOnce(); await w.basisOfRound();
|
||||
const sent = await w.mouth.send({ requestId: "r1", persona: "eureka", roomAlias: ALIAS, text: "hello" });
|
||||
assert.equal(sent.refusal, null);
|
||||
w.at(T0 + 8 * MIN);
|
||||
await w.basisOfRound(); // the echo, 8 minutes on: a second anchor would carry the window to 18 minutes
|
||||
w.at(T0 + 12 * MIN);
|
||||
assert.deepEqual(await w.basisOfRound(), [["$late", null]]);
|
||||
});
|
||||
|
||||
test("follow: a send from the mouth anchors; a send into a DM or an unreadable room does not", async () => {
|
||||
const mk = opts => world({ ...opts, rounds: [{ [ROOM]: [msg("$k0", KIKO, T0 - MIN)] }, { [ROOM]: [msg("$k1", KIKO, T0 + MIN)] }] });
|
||||
const ok = mk({});
|
||||
await ok.mouth.syncOnce(); await ok.basisOfRound();
|
||||
await ok.mouth.send({ requestId: "r1", persona: "eureka", roomAlias: ALIAS, text: "hello" });
|
||||
assert.deepEqual(await ok.basisOfRound(), [["$k1", "follow"]]);
|
||||
const dm = mk({ direct: { [KIKO]: [ROOM] } });
|
||||
await dm.mouth.syncOnce(); await dm.basisOfRound();
|
||||
await dm.mouth.send({ requestId: "r1", persona: "eureka", roomAlias: ROOM, text: "hello" });
|
||||
assert.deepEqual(await dm.basisOfRound(), [["$k1", "direct"]]);
|
||||
});
|
||||
|
||||
test("follow: a DM, an encrypted room and a bridged room anchor nothing; an unreadable or non-array state is not public", async () => {
|
||||
const rounds = () => [{ [ROOM]: [msg("$k0", KIKO, T0 - MIN)] }, { [ROOM]: [mine("$mine", T0)] }, { [ROOM]: [msg("$k", KIKO, T0 + MIN)] }];
|
||||
for (const state of [[{ type: "m.room.encryption", content: {} }], [{ type: "m.bridge", content: {} }], [{ type: "uk.half-shot.bridge", content: {} }], { not: "an array" }]) {
|
||||
assert.deepEqual((await run(world({ state: { [ROOM]: state }, rounds: rounds() }), 3))[2], [["$k", null]]);
|
||||
}
|
||||
assert.deepEqual((await run(world({ stateStatus: { [ROOM]: 500 }, rounds: rounds() }), 3))[2], [["$k", null]]);
|
||||
assert.deepEqual((await run(world({ direct: { [KIKO]: [ROOM] }, rounds: rounds() }), 3))[2], [["$k", "direct"]]);
|
||||
// plaintext and unbridged is public, whatever else it is (an invite-only room included): the state holds no join rule the mouth reads
|
||||
assert.deepEqual((await run(world({ state: { [ROOM]: [{ type: "m.room.join_rules", content: { join_rule: "invite" } }] }, rounds: rounds() }), 3))[2], [["$k", "follow"]]);
|
||||
});
|
||||
|
||||
test("follow: a bridged sender or event is neither counted nor followed", async () => {
|
||||
const puppet = `@_${PLATFORM}_123:${HS}`;
|
||||
const w = world({ rounds: [{ [ROOM]: [msg("$p0", puppet, T0 - MIN), msg("$k0", KIKO, T0 - MIN)] }, { [ROOM]: [mine("$mine", T0, { "m.mentions": { user_ids: [puppet] } })] },
|
||||
{ [ROOM]: [msg("$p", puppet, T0 + MIN), msg("$k", KIKO, T0 + MIN, { [`${PLATFORM}.relay`]: true })] }] });
|
||||
assert.deepEqual((await run(w, 3))[2].sort(), [["$k", null], ["$p", null]]);
|
||||
});
|
||||
|
||||
test("follow: at most three an hour per room and peer, counted independently; a half hour does not reset it; an hour later it is follow again", async () => {
|
||||
const w = world({ rounds: [
|
||||
{ [ROOM]: [msg("$k0", KIKO, T0 - SEC), msg("$l0", LENA, T0 - SEC)], [OTHER_ROOM]: [msg("$ko0", KIKO, T0 - SEC)] },
|
||||
{ [ROOM]: [mine("$m1", T0 + MIN)], [OTHER_ROOM]: [msg("$mo1", ME, T0 + MIN)] },
|
||||
{ [ROOM]: [msg("$k1", KIKO, T0 + MIN + SEC)], [OTHER_ROOM]: [msg("$ko1", KIKO, T0 + MIN + SEC)] },
|
||||
{ [ROOM]: [mine("$m2", T0 + 2 * MIN)] }, { [ROOM]: [msg("$k2", KIKO, T0 + 2 * MIN + SEC)] },
|
||||
{ [ROOM]: [mine("$m3", T0 + 3 * MIN)] }, { [ROOM]: [msg("$k3", KIKO, T0 + 3 * MIN + SEC)] },
|
||||
{ [ROOM]: [mine("$m4", T0 + 4 * MIN)] }, { [ROOM]: [msg("$k4", KIKO, T0 + 4 * MIN + SEC), msg("$l4", LENA, T0 + 4 * MIN + SEC)] },
|
||||
{ [ROOM]: [msg("$k30", KIKO, T0 + 30 * MIN), mine("$m31", T0 + 31 * MIN)] }, { [ROOM]: [msg("$k31", KIKO, T0 + 31 * MIN + SEC)] },
|
||||
{ [ROOM]: [msg("$k60", KIKO, T0 + 61 * MIN), mine("$m62", T0 + 62 * MIN)] }, { [ROOM]: [msg("$k62", KIKO, T0 + 62 * MIN + SEC)] },
|
||||
] });
|
||||
await w.mouth.syncOnce();
|
||||
const seen = [];
|
||||
const steps = [0, 1, 1, 2, 2, 3, 3, 4, 4, 31, 31, 62, 62];
|
||||
for (const m of steps) seen.push(...await go(w, T0 + m * MIN + 2 * SEC));
|
||||
const got = Object.fromEntries(seen.filter(([id]) => /^\$(k|ko|l)\d/.test(id)).map(([id, b]) => [id, b]));
|
||||
assert.deepEqual(got, { $k0: null, $l0: null, $ko0: null, $k1: "follow", $ko1: "follow", $k2: "follow", $k3: "follow", $k4: null, $l4: "follow", $k30: null, $k31: null, $k60: null, $k62: "follow" });
|
||||
});
|
||||
|
||||
test("follow: an edit and a notice are never follow; a room-basis message from a counted sender is; a tag keeps its own basis", async () => {
|
||||
const w = world({ rounds: [
|
||||
{ [ROOM]: [msg("$k0", KIKO, T0 - MIN)] },
|
||||
{ [ROOM]: [mine("$mine", T0)] },
|
||||
{ [ROOM]: [
|
||||
msg("$edit", KIKO, T0 + SEC, { "m.new_content": { msgtype: "m.text", body: "x" }, "m.relates_to": { rel_type: "m.replace", event_id: "$k0" } }),
|
||||
msg("$notice", KIKO, T0 + SEC, { msgtype: "m.notice" }),
|
||||
msg("$room", KIKO, T0 + SEC, { "m.mentions": { room: true } }),
|
||||
msg("$tag", KIKO, T0 + SEC, { "m.mentions": { user_ids: [ME] } }),
|
||||
] },
|
||||
] });
|
||||
assert.deepEqual((await run(w, 3))[2].sort(), [["$edit", null], ["$notice", null], ["$room", "follow"], ["$tag", "tag"]]);
|
||||
});
|
||||
|
||||
test("follow: flag off emits nothing new and reads nothing extra; a fresh mouth has no engagement", async () => {
|
||||
const rounds = [{ [ROOM]: [msg("$k0", KIKO, T0 - MIN)] }, { [ROOM]: [mine("$mine", T0)] }, { [ROOM]: [msg("$k", KIKO, T0 + MIN)] }];
|
||||
const off = world({ follow: false, rounds });
|
||||
assert.deepEqual((await run(off, 3))[2], [["$k", null]]);
|
||||
assert.ok(!off.seen.some(l => l.endsWith("/state")), "off: no state read");
|
||||
const restarted = world({ rounds: [{ [ROOM]: [msg("$k", KIKO, T0 + MIN)] }] });
|
||||
assert.deepEqual((await run(restarted, 1))[0], [["$k", null]]);
|
||||
assert.equal(followFromEnv({ CHAT_FOLLOW: "1" }), true);
|
||||
for (const v of [undefined, "0", "true", "yes", ""]) assert.equal(followFromEnv({ CHAT_FOLLOW: v }), false);
|
||||
});
|
||||
|
||||
test("follow: the basis is a name the delivery document accepts", () => {
|
||||
assert.ok(docs.bases.includes("follow"));
|
||||
const d = { seq: 1, roomAlias: ALIAS, roomId: ROOM, eventId: "$e", sender: KIKO, senderDisplay: "kiko", text: "x", atUnixMs: 1, context: [], image: null, mentions: [], isDirect: false, basis: "follow", form: "message" };
|
||||
assert.equal(docs.parseDeliveryV3(docs.deliveryV3Tuple(d)).basis, "follow");
|
||||
});
|
||||
|
||||
const tick = async () => { for (let i = 0; i < 20; i++) await new Promise(r => setImmediate(r)); };
|
||||
|
||||
test("follow: a send receipt is not delayed by a room-state read that hangs or fails; the anchor happens in the background", async () => {
|
||||
for (const opts of [{ hangState: true }, { stateStatus: { [ROOM]: 500 } }]) {
|
||||
const w = world({ ...opts, rounds: [] });
|
||||
await w.mouth.syncOnce();
|
||||
const sent = await Promise.race([w.mouth.send({ requestId: "r1", persona: "eureka", roomAlias: ALIAS, text: "hello" }), new Promise(r => setTimeout(() => r("late"), 500))]);
|
||||
assert.equal(sent.refusal, null);
|
||||
}
|
||||
const ok = world({ rounds: [{ [ROOM]: [msg("$k0", KIKO, T0 - MIN)] }, { [ROOM]: [msg("$k1", KIKO, T0 + MIN)] }] });
|
||||
await ok.mouth.syncOnce(); await ok.basisOfRound();
|
||||
await ok.mouth.send({ requestId: "r1", persona: "eureka", roomAlias: ALIAS, text: "hello" });
|
||||
await tick();
|
||||
assert.deepEqual(await ok.basisOfRound(), [["$k1", "follow"]]);
|
||||
});
|
||||
|
||||
test("follow: the flag off makes a send cost no extra call and no extra turn", async () => {
|
||||
const w = world({ follow: false, rounds: [] });
|
||||
await w.mouth.syncOnce();
|
||||
const before = w.seen.length;
|
||||
await w.mouth.send({ requestId: "r1", persona: "eureka", roomAlias: ALIAS, text: "hello" });
|
||||
await tick();
|
||||
assert.ok(!w.seen.slice(before).some(l => l.endsWith("/state")));
|
||||
});
|
||||
|
||||
test("follow: a homeserver clock up to 60 s behind still follows; 61 s early does not; the very end of the window is out, 1 ms before it is in", async () => {
|
||||
const w = world({ rounds: [
|
||||
{ [ROOM]: [msg("$k0", KIKO, T0 - MIN), msg("$l0", LENA, T0 - MIN), msg("$e0", `@e:${HS}`, T0 - MIN), msg("$f0", `@f:${HS}`, T0 - MIN)] },
|
||||
{ [ROOM]: [mine("$mine", T0)] },
|
||||
{ [ROOM]: [msg("$x1", KIKO, T0 - 60 * SEC), msg("$x2", LENA, T0 - 61 * SEC), msg("$x3", `@e:${HS}`, T0 + W - 1), msg("$x4", `@f:${HS}`, T0 + W)] },
|
||||
] });
|
||||
await w.mouth.syncOnce(); await go(w, T0 - MIN); await go(w, T0);
|
||||
assert.deepEqual((await go(w, T0 + 5 * SEC)).sort(), [["$x1", "follow"], ["$x2", null], ["$x3", "follow"], ["$x4", null]]);
|
||||
});
|
||||
|
||||
test("follow: a window that ended while the pass was being read is out, whatever the message says", async () => {
|
||||
let jump = null;
|
||||
const w = world({ onSync: set => { if (jump !== null) set(jump); }, rounds: [{ [ROOM]: [msg("$k0", KIKO, T0 - MIN)] }, { [ROOM]: [mine("$mine", T0)] }, { [ROOM]: [msg("$x", KIKO, T0 + MIN)] }] });
|
||||
await w.mouth.syncOnce(); await go(w, T0 - MIN);
|
||||
const keep = w.mouth.followState;
|
||||
w.at(T0); await w.basisOfRound(); // her post; the next sync moves the clock after the purge
|
||||
assert.equal(keep().engagements.length, 1);
|
||||
jump = T0 + 30 * MIN; w.at(T0); // the purge at the start of the pass sees T0; the clock moves while the pass reads
|
||||
assert.deepEqual(await w.basisOfRound(), [["$x", null]]);
|
||||
});
|
||||
|
||||
test("follow: the cap counts only the hour before the message", async () => {
|
||||
const w = world({ rounds: [
|
||||
{ [ROOM]: [msg("$k0", KIKO, T0 - SEC)] },
|
||||
...[1, 2, 3].flatMap(n => [{ [ROOM]: [mine(`$m${n}`, T0 + n * MIN)] }, { [ROOM]: [msg(`$x${n}`, KIKO, T0 + n * MIN + SEC)] }]),
|
||||
{ [ROOM]: [msg("$k90", KIKO, T0 + 89 * MIN), mine("$m90", T0 + 90 * MIN)] }, { [ROOM]: [msg("$x90", KIKO, T0 + 90 * MIN + SEC)] },
|
||||
] });
|
||||
await w.mouth.syncOnce(); await go(w, T0);
|
||||
for (const n of [1, 1, 2, 2, 3, 3]) await go(w, T0 + n * MIN + SEC);
|
||||
assert.deepEqual(await go(w, T0 + 30 * MIN), [["$k90", null]]); // stamps run ahead of this process's clock, so only the cap's own hour bound can tell
|
||||
assert.deepEqual(await go(w, T0 + 30 * MIN), [["$x90", "follow"]]);
|
||||
});
|
||||
|
||||
test("follow: whoever her reply quotes is counted though they spoke long ago; a thread fallback pointer names no one", async () => {
|
||||
const w = world({ rounds: [
|
||||
{ [ROOM]: [msg("$old", LENA, T0 - 30 * MIN), msg("$ghost", `@ghost:${HS}`, T0 - 30 * MIN)] },
|
||||
{ [ROOM]: [mine("$mine", T0, { "m.relates_to": { "m.in_reply_to": { event_id: "$old" } } })] },
|
||||
{ [ROOM]: [msg("$x1", LENA, T0 + MIN)] },
|
||||
{ [ROOM]: [msg("$old2", KIKO, T0 + 2 * MIN)] },
|
||||
{ [ROOM]: [mine("$mine2", T0 + 3 * MIN, { "m.relates_to": { rel_type: "m.thread", event_id: "$old", is_falling_back: true, "m.in_reply_to": { event_id: "$ghost" } } })] },
|
||||
{ [ROOM]: [msg("$x2", STRANGER, T0 + 4 * MIN)] },
|
||||
] });
|
||||
await w.mouth.syncOnce();
|
||||
await go(w, T0 - 30 * MIN); await go(w, T0);
|
||||
assert.deepEqual(await go(w, T0 + MIN), [["$x1", "follow"]]);
|
||||
await go(w, T0 + 2 * MIN);
|
||||
await go(w, T0 + 3 * MIN);
|
||||
const [engagement] = w.mouth.followState().engagements;
|
||||
assert.ok(engagement.counted.includes(KIKO), "kiko spoke 1 minute before the second post");
|
||||
assert.ok(!engagement.counted.includes(`@ghost:${HS}`), "a fallback pointer names no one");
|
||||
});
|
||||
|
||||
test("follow: someone stamped after her send is not counted as having spoken before it", async () => {
|
||||
const w = world({ rounds: [{ [ROOM]: [msg("$ahead", KIKO, T0 + MIN)] }, { [ROOM]: [msg("$x", KIKO, T0 + 2 * MIN)] }] });
|
||||
await w.mouth.syncOnce(); await w.basisOfRound();
|
||||
await w.mouth.send({ requestId: "r1", persona: "eureka", roomAlias: ALIAS, text: "hello" });
|
||||
await tick();
|
||||
assert.deepEqual(w.mouth.followState().engagements[0].counted, []);
|
||||
assert.deepEqual(await go(w, T0 + 2 * MIN), [["$x", null]]);
|
||||
});
|
||||
|
||||
test("follow: the ring keeps the newest 200 senders of a room; a flood pushes out the oldest, not the newest", async () => {
|
||||
const flood = Array.from({ length: 250 }, (_, i) => msg(`$f${i}`, `@u${i}:${HS}`, T0 - 5 * MIN + i));
|
||||
const w = world({ rounds: [{ [ROOM]: flood }, { [ROOM]: [mine("$mine", T0)] }, { [ROOM]: [msg("$x0", `@u0:${HS}`, T0 + MIN), msg("$x249", `@u249:${HS}`, T0 + MIN)] }] });
|
||||
await w.mouth.syncOnce(); await w.basisOfRound();
|
||||
assert.equal(w.mouth.followState().spoke[0].senders.length, 200);
|
||||
await go(w, T0);
|
||||
assert.deepEqual(await go(w, T0 + MIN + SEC), [["$x0", null], ["$x249", "follow"]]);
|
||||
assert.equal(w.mouth.followState().engagements[0].counted.length, 200);
|
||||
});
|
||||
|
||||
test("follow: only senders, and only real ones, are remembered and counted: not her, not a bridged id, not an edit, not a notice, not a malformed id", async () => {
|
||||
const puppet = `@_${PLATFORM}_9:${HS}`;
|
||||
const w = world({ rounds: [
|
||||
{ [ROOM]: [msg("$a", KIKO, T0 - 2 * MIN), mine("$self", T0 - 2 * MIN), msg("$b", puppet, T0 - 2 * MIN),
|
||||
msg("$c", LENA, T0 - 2 * MIN, { "m.relates_to": { rel_type: "m.replace", event_id: "$a" } }), msg("$d", STRANGER, T0 - 2 * MIN, { msgtype: "m.notice" })] },
|
||||
{ [ROOM]: [mine("$mine", T0, { "m.mentions": { user_ids: [ME, puppet, "not-an-id", `@${"x".repeat(300)}:${HS}`, LENA] } })] },
|
||||
] });
|
||||
await w.mouth.syncOnce(); await go(w, T0 - MIN);
|
||||
assert.deepEqual(w.mouth.followState().spoke, [{ room: ROOM, senders: [KIKO] }]);
|
||||
await go(w, T0);
|
||||
assert.deepEqual(w.mouth.followState().engagements[0].counted.sort(), [KIKO, LENA].sort());
|
||||
});
|
||||
|
||||
test("follow: her own post in the baseline sync, which is history, anchors nothing", async () => {
|
||||
const w = world({ baselineEvents: { [ROOM]: [msg("$k0", KIKO, T0 - MIN), mine("$mine", T0)] }, rounds: [] });
|
||||
await w.mouth.syncOnce();
|
||||
assert.deepEqual(w.mouth.followState().engagements, []);
|
||||
});
|
||||
|
||||
test("follow: every pass forgets what has aged out: anchors, speakers and cap times", async () => {
|
||||
const w = world({ rounds: [
|
||||
{ [ROOM]: [msg("$k0", KIKO, T0 - MIN)] }, { [ROOM]: [mine("$mine", T0)] }, { [ROOM]: [msg("$x", KIKO, T0 + MIN)] }, {},
|
||||
] });
|
||||
await w.mouth.syncOnce(); await go(w, T0 - MIN); await go(w, T0); await go(w, T0 + MIN + SEC);
|
||||
const s1 = w.mouth.followState();
|
||||
assert.equal(s1.engagements.length, 1); assert.equal(s1.followsGiven, 1); assert.equal(s1.spoke.length, 1);
|
||||
await go(w, T0 + 3 * 60 * MIN);
|
||||
assert.deepEqual(w.mouth.followState(), { engagements: [], spoke: [], followsGiven: 0 });
|
||||
});
|
||||
|
||||
test("follow: a failed room-state read at send time leaves the post retryable: its echo anchors it", async () => {
|
||||
const w = world({ stateFailures: 1, rounds: [{ [ROOM]: [msg("$k0", KIKO, T0 - MIN)] }, { [ROOM]: [msg("$sent1", ME, T0, { body: "hello" })] }, { [ROOM]: [msg("$x", KIKO, T0 + MIN)] }] });
|
||||
await w.mouth.syncOnce(); await w.basisOfRound();
|
||||
await w.mouth.send({ requestId: "r1", persona: "eureka", roomAlias: ALIAS, text: "hello" });
|
||||
await tick();
|
||||
assert.deepEqual(w.mouth.followState().engagements, [], "the read failed: nothing anchored yet");
|
||||
await go(w, T0);
|
||||
assert.equal(w.mouth.followState().engagements.length, 1, "the echo anchored it");
|
||||
assert.deepEqual(await go(w, T0 + MIN), [["$x", "follow"]]);
|
||||
});
|
||||
|
||||
test("follow: a send and its echo in flight together anchor once, and a failure of that one read leaves the echo to retry", async () => {
|
||||
const w = world({ stateFailures: 1, rounds: [{ [ROOM]: [msg("$k0", KIKO, T0 - MIN)] }, { [ROOM]: [msg("$sent1", ME, T0, { body: "hello" })] }] });
|
||||
await w.mouth.syncOnce(); await w.basisOfRound();
|
||||
const sending = w.mouth.send({ requestId: "r1", persona: "eureka", roomAlias: ALIAS, text: "hello" });
|
||||
await sending;
|
||||
await w.basisOfRound(); // the echo, possibly while the send's read is still in flight
|
||||
await tick();
|
||||
assert.equal(w.mouth.followState().engagements.length, 1);
|
||||
assert.equal(w.seen.filter(l => l.endsWith("/state")).length, 2, "one failed read, one that worked; no third");
|
||||
});
|
||||
|
||||
test("follow: a not-public verdict is recorded: the room is not read again for each post or pass", async () => {
|
||||
const w = world({ state: { [ROOM]: [{ type: "m.bridge", content: {} }] }, rounds: [
|
||||
{ [ROOM]: [msg("$k0", KIKO, T0 - MIN)] }, { [ROOM]: [msg("$sent1", ME, T0)] }, { [ROOM]: [mine("$a", T0 + SEC), msg("$x", KIKO, T0 + 2 * SEC)] }, { [ROOM]: [mine("$b", T0 + 3 * SEC)] },
|
||||
] });
|
||||
await w.mouth.syncOnce(); await w.basisOfRound();
|
||||
await w.mouth.send({ requestId: "r1", persona: "eureka", roomAlias: ALIAS, text: "hello" });
|
||||
await tick();
|
||||
for (const t of [0, 0, 0]) await go(w, T0 + t);
|
||||
assert.equal(w.seen.filter(l => l.endsWith("/state")).length, 1);
|
||||
assert.deepEqual(w.mouth.followState().engagements, []);
|
||||
});
|
||||
|
||||
test("follow: an echo that arrives while the send's anchor is still reading waits for it: one read, one snapshot", async () => {
|
||||
const w = world({ stateDelay: 80, rounds: [{ [ROOM]: [msg("$k0", KIKO, T0 - MIN)] }, { [ROOM]: [msg("$sent1", ME, T0 + 5 * SEC, { body: "hello" })] }] });
|
||||
await w.mouth.syncOnce(); await w.basisOfRound();
|
||||
await w.mouth.send({ requestId: "r1", persona: "eureka", roomAlias: ALIAS, text: "hello" }); // its anchor is now waiting on the state read
|
||||
await w.basisOfRound(); // the echo, meanwhile
|
||||
await new Promise(r => setTimeout(r, 200));
|
||||
assert.equal(w.seen.filter(l => l.endsWith("/state")).length, 1);
|
||||
assert.equal(w.mouth.followState().engagements[0].from, T0, "anchored at the send's time, not the echo's");
|
||||
});
|
||||
|
|
@ -68,8 +68,8 @@ export function parsePendingRequest(t) {
|
|||
|
||||
// What the mouth can establish about a delivery from Matrix alone, for the Persona it is delivered to (see docs/bridge.md, "Addressing facts").
|
||||
// `basis` is why the message is for her, or null when the mouth has no proof (not that it is not for her: her names, her projects' names and
|
||||
// the sender's role are not the mouth's to know). `form` is the event's kind: only a "message" can address anyone.
|
||||
export const bases = Object.freeze(["direct", "tag", "reply", "room"]);
|
||||
// the sender's role are not the mouth's to know). "follow" is emitted only under the mouth's CHAT_FOLLOW flag. `form` is the event's kind: only a "message" can address anyone.
|
||||
export const bases = Object.freeze(["direct", "tag", "reply", "room", "follow"]);
|
||||
export const forms = Object.freeze(["message", "notice", "edit"]);
|
||||
|
||||
export function ackTuple({ persona, throughSeq }) {
|
||||
|
|
|
|||
|
|
@ -6,7 +6,7 @@
|
|||
// its m.mentions user ids and whether the room is a DM. The /sync position
|
||||
// (next_batch) is persisted only when an ack covers it, so a restart before the
|
||||
// ack re-delivers instead of losing. A v3 delivery also carries what Matrix proves about why it is for her (basis: direct, tag,
|
||||
// reply, room; form: message, notice, edit); names and roles are the heartbeat's to match, not the mouth's (docs/bridge.md,
|
||||
// reply, room, and with CHAT_FOLLOW=1 follow; form: message, notice, edit); names and roles are the heartbeat's to match, not the mouth's (docs/bridge.md,
|
||||
// "Addressing facts"). Reading wakes nothing: the heartbeat polls
|
||||
// with a non-consuming chat_pending_request and decides who runs. An ack posts nothing to Matrix. Her public
|
||||
// m.read receipt follows a chat_read document from Huginn, sent when a delivery has gone into the model's
|
||||
|
|
@ -23,6 +23,11 @@
|
|||
// reads the room's head from Matrix and answers RoomMoved, naming the messages that moved it, instead
|
||||
// of the PUT. A v1 send, and a v2 send with no basis, are not checked; both still pass through the gate.
|
||||
//
|
||||
// Follow (CHAT_FOLLOW=1, default off; docs/bridge.md "Follow"): her own send in a public room (not a DM, not encrypted, not bridged) anchors the room
|
||||
// for 10 minutes; those who spoke there in the 10 minutes before it, and whoever it tagged or replied to, get basis "follow" on their later messages
|
||||
// inside the window, at most 3 an hour per peer and room. Memory only. Off, no basis "follow" is ever emitted: Huginn's parser refuses it until its
|
||||
// Basis::Follow lands.
|
||||
//
|
||||
// Media: an m.image/m.file/m.audio/m.video event is delivered as one typed line (see bodyText), not its bare filename.
|
||||
// An m.image event in a delivered message is also fetched, as this Persona and only when an image store is configured
|
||||
// (persona-chat-images.mjs): the delivery then carries a reference to the stored file, or the line says why not.
|
||||
|
|
@ -68,6 +73,12 @@ const FETCH_DEADLINE_MS = 15000; // one fetch, whole: it ends before Huginn's 20
|
|||
const FETCH_PER_HOUR = 30; // fetches one Persona may ask in an hour
|
||||
const RELATION_PAGES = 3; // pages of a receipt target's relations read before the answer is called unproven
|
||||
const PLATFORM = ["disc", "ord"].join(""); // the bridged platform's name, as Huginn's room_scope spells it: never written out whole
|
||||
const FOLLOW_WINDOW_MS = 10 * 60 * 1000; // W: an engagement lasts this long from the Mind's own post; a follow never extends it
|
||||
const FOLLOW_PER_HOUR = 3; // K: follow bases given per (Mind, peer, room) in an hour; past it the message keeps the basis it had
|
||||
const FOLLOW_ROOM_TTL_MS = 5 * 60 * 1000; // how long a room's public verdict (not encrypted, not bridged) is kept
|
||||
const FOLLOW_SLACK_MS = 60 * 1000; // clock disagreement tolerated between this process and the homeserver
|
||||
const FOLLOW_HOUR_MS = 60 * 60 * 1000;
|
||||
const FOLLOW_RING = 200; // recent senders kept per room
|
||||
const HEAD_PAGE = 50; // the room head a gated send reads; also the most ids a RoomMoved receipt names
|
||||
|
||||
// The kind of an event for a delivery: an edit (m.replace) and an automated notice (m.notice) are not messages anyone is addressed by.
|
||||
|
|
@ -126,6 +137,7 @@ export function createMouth({
|
|||
persona, homeserver, serverName, rooms, token, fetch: httpFetch = globalThis.fetch, state, brake,
|
||||
now = Date.now, log = () => {}, syncTimeoutMs = 30000, maxPending = MAX_PENDING, gate = createRoomGate(), sendTimeoutMs = SEND_TIMEOUT_MS,
|
||||
presenceGraceMs = PRESENCE_GRACE_MS, fetchDeadlineMs = FETCH_DEADLINE_MS,
|
||||
follow = false, // the follow basis (CHAT_FOLLOW): an untagged message from someone she tagged or replied to, within FOLLOW_WINDOW_MS of her own post, is for her
|
||||
replyLink = false, // resolve each reply's quoted event into delivery.replyTo (the v4 answer; one more event read per reply to a non-Mind)
|
||||
images = null, // this Persona's image store (createImageStore), or null: images stay the typed text line
|
||||
isServed = async () => false, // (async) whether a Matrix user id is an account this process speaks for: a Mind is not a person to open a DM with
|
||||
|
|
@ -303,6 +315,96 @@ export function createMouth({
|
|||
return lines;
|
||||
}
|
||||
|
||||
// Follow (docs/bridge.md, "Follow"). Memory only, bounded, purged on every sync pass. Her own send in a public room (not a DM, not encrypted, not
|
||||
// bridged) anchors that room for W from the send: `engagements` room id -> { from, ends, counted }, counted being the senders who spoke there in the
|
||||
// W before the send (`spoke`, a per-room ring of recent senders with event times) plus whoever the post tagged or replied to, if the event says so.
|
||||
// Someone who first speaks after the send is not counted. Only her speaking again re-anchors (and re-snapshots); a follow never extends the window.
|
||||
// `followsGiven` room|peer -> times follows were given. A restart empties all of it, which only shortens windows.
|
||||
const engagements = new Map();
|
||||
const spoke = new Map();
|
||||
const followsGiven = new Map();
|
||||
const anchored = new Set(); // event ids of her own posts already anchored: the send path and its /sync echo anchor once
|
||||
const anchoring = new Map(); // event ids whose anchor is in flight: the send path and its echo never snapshot twice
|
||||
const publicRooms = new Map(); // room id -> { ok, at }: the verdicts, a "no" included, so a not-public room is read once per FOLLOW_ROOM_TTL_MS, not every pass
|
||||
const platformId = id => platformEvent({ sender: id });
|
||||
// true or false is a verdict (kept); null is a failed read: unknown, retryable, kept nowhere.
|
||||
async function publicRoom(roomId) {
|
||||
if (directRooms.has(roomId)) return false;
|
||||
const known = publicRooms.get(roomId);
|
||||
if (known && now() - known.at < FOLLOW_ROOM_TTL_MS) return known.ok;
|
||||
try {
|
||||
const state = await matrix("GET", `/rooms/${encodeURIComponent(roomId)}/state`, undefined, API, AbortSignal.timeout(sendTimeoutMs));
|
||||
if (!Array.isArray(state)) return null;
|
||||
const ok = !state.some(e => e?.type === "m.room.encryption" || e?.type === "m.bridge" || e?.type === "uk.half-shot.bridge");
|
||||
publicRooms.set(roomId, { ok, at: now() });
|
||||
return ok;
|
||||
} catch (error) {
|
||||
say(`follow: room ${roomId} state not read: ${error.message}`);
|
||||
return null; // unknown is not public, and not a verdict either
|
||||
}
|
||||
}
|
||||
const eventAt = event => Number.isSafeInteger(event.origin_server_ts) && event.origin_server_ts >= 0 ? event.origin_server_ts : now();
|
||||
function purgeFollow() {
|
||||
const t = now();
|
||||
for (const [room, e] of engagements) if (e.ends + FOLLOW_SLACK_MS <= t) engagements.delete(room);
|
||||
for (const [room, ring] of spoke) {
|
||||
const kept = ring.filter(r => r.at > t - FOLLOW_WINDOW_MS - FOLLOW_SLACK_MS);
|
||||
if (kept.length === 0) spoke.delete(room); else spoke.set(room, kept);
|
||||
}
|
||||
for (const [key, times] of followsGiven) {
|
||||
const kept = times.filter(x => t - x < FOLLOW_HOUR_MS + FOLLOW_SLACK_MS);
|
||||
if (kept.length === 0) followsGiven.delete(key); else followsGiven.set(key, kept);
|
||||
}
|
||||
}
|
||||
// Someone spoke in the room (a message from another account, not an edit or a notice): the ring behind the snapshot.
|
||||
function noteSpeaker(roomId, event) {
|
||||
if (!follow || directRooms.has(roomId) || formOf(event.content) !== "message" || platformEvent(event) || event.sender === selfId) return;
|
||||
const ring = spoke.get(roomId) ?? [];
|
||||
ring.push({ sender: event.sender, at: eventAt(event) });
|
||||
if (ring.length > FOLLOW_RING) ring.shift();
|
||||
spoke.set(roomId, ring);
|
||||
}
|
||||
// Her post at time `at` (the send's own time, or the echo's origin time): snapshot the room. `tagged`: ids from the event's m.mentions, `replyTo` the quoted event id.
|
||||
async function anchor(roomId, eventId, at, tagged = [], replyTo = null) {
|
||||
if (!follow) return;
|
||||
while (anchoring.has(eventId)) await anchoring.get(eventId).catch(() => {}); // the send and its echo meet here: the second waits, then sees whether the first succeeded
|
||||
if (anchored.has(eventId)) return;
|
||||
const run = (async () => {
|
||||
const verdict = await publicRoom(roomId);
|
||||
if (verdict === null) return; // the read failed: this post stays retryable, its echo tries again
|
||||
anchored.add(eventId);
|
||||
if (anchored.size > 200) anchored.delete(anchored.values().next().value);
|
||||
if (!verdict) return; // a "no" stays recorded
|
||||
const counted = new Set((spoke.get(roomId) ?? []).filter(r => r.at >= at - FOLLOW_WINDOW_MS && r.at <= at).map(r => r.sender));
|
||||
for (const id of tagged) counted.add(id);
|
||||
if (typeof replyTo === "string" && replyTo !== "") { const author = await authorOf(roomId, replyTo); if (author !== null) counted.add(author); }
|
||||
for (const id of [...counted]) if (!docs.isMatrixId("@", id) || id === selfId || platformId(id)) counted.delete(id);
|
||||
engagements.set(roomId, { from: at, ends: at + FOLLOW_WINDOW_MS, counted });
|
||||
})();
|
||||
anchoring.set(eventId, run);
|
||||
try { await run; } finally { anchoring.delete(eventId); }
|
||||
}
|
||||
// Her own event seen in /sync (a send from another client, or the echo of a mouth send, which is then not anchored twice).
|
||||
async function noteOwnPost(roomId, event) {
|
||||
if (!follow || formOf(event.content) !== "message" || platformEvent(event)) return;
|
||||
const relates = event.content["m.relates_to"];
|
||||
const tagged = Array.isArray(event.content["m.mentions"]?.user_ids) ? event.content["m.mentions"].user_ids : [];
|
||||
await anchor(roomId, event.event_id, eventAt(event), tagged.slice(0, docs.limits.mentions), relates?.is_falling_back === true ? null : relates?.["m.in_reply_to"]?.event_id);
|
||||
}
|
||||
// Whether this message is a follow, and if so spends one of the peer's K for the hour. Called only for a message that has no better basis.
|
||||
async function followed(roomId, event) {
|
||||
if (!follow || directRooms.has(roomId) || formOf(event.content) !== "message" || platformEvent(event)) return false;
|
||||
const e = engagements.get(roomId), at = eventAt(event), t = now();
|
||||
// Both clocks must agree: the message's own time inside the window, and this process's time not past it (a forged or old timestamp reuses nothing).
|
||||
if (!e || !e.counted.has(event.sender) || at < e.from - FOLLOW_SLACK_MS || at >= e.ends || t >= e.ends + FOLLOW_SLACK_MS) return false;
|
||||
if (!(await publicRoom(roomId))) return false;
|
||||
const key = `${roomId}|${event.sender}`;
|
||||
const recent = (followsGiven.get(key) ?? []).filter(x => at - x < FOLLOW_HOUR_MS);
|
||||
if (recent.length >= FOLLOW_PER_HOUR) { followsGiven.set(key, recent); return false; }
|
||||
followsGiven.set(key, [...recent, at]);
|
||||
return true;
|
||||
}
|
||||
|
||||
// Why a delivered event is for this Persona, from what Matrix says and nothing else: "direct" (a DM), "tag" (m.mentions lists her
|
||||
// id, or, from a client that sends no m.mentions at all, the body links her matrix.to pill), "reply" (it replies to, or sits in a
|
||||
// thread under, an event her account wrote, so her Self run's words count as hers), "room" (an @room mention). In that order; null
|
||||
|
|
@ -440,6 +542,7 @@ export function createMouth({
|
|||
// One /sync round. Returns the deliveries it made.
|
||||
async function syncOnce() {
|
||||
if (!started || unjoined()) await start();
|
||||
purgeFollow();
|
||||
replyReads = 0;
|
||||
held.clear();
|
||||
const free = await released(); // the round's brake read for presence, taken before any early return
|
||||
|
|
@ -474,7 +577,10 @@ export function createMouth({
|
|||
const deliverable = !baseline && !freshRooms.has(roomId) && event.sender !== selfId;
|
||||
if (deliverable && pending.length + made.length >= maxPending) { truncated = true; break rounds; }
|
||||
if (deliverable && replyLink && replyReads + targetsToRead(roomId, event) > REPLY_READS_PER_ROUND) { say(`event ${event.event_id} not read: round cap on target reads, the rest of the round waits for the next`); truncated = true; break rounds; } // not processed: the next round meets it again, with its read count reset
|
||||
const basis = deliverable ? await basisOf(roomId, event) : null;
|
||||
if (!baseline && !freshRooms.has(roomId) && event.sender === selfId) await noteOwnPost(roomId, event);
|
||||
noteSpeaker(roomId, event);
|
||||
let basis = deliverable ? await basisOf(roomId, event) : null;
|
||||
if (deliverable && (basis === null || basis === "room") && await followed(roomId, event)) basis = "follow";
|
||||
const replyTo = deliverable ? await replyToOf(roomId, event) : null;
|
||||
processed.add(event.event_id);
|
||||
if (deliverable) {
|
||||
|
|
@ -647,7 +753,10 @@ export function createMouth({
|
|||
const sent = await matrix("PUT", `/rooms/${encodeURIComponent(roomId)}/send/m.room.message/${encodeURIComponent(requestId)}`, { msgtype: "m.text", body: text }, API, AbortSignal.timeout(sendTimeoutMs));
|
||||
return { requestId, eventId: sent.event_id, refusal: null, newer: [] };
|
||||
});
|
||||
if (result.refusal === null) { touch(); setPresence(true).catch(() => {}); } // outside the room gate and not awaited: a slow presence write holds neither another send nor this receipt
|
||||
if (result.refusal === null) {
|
||||
touch(); setPresence(true).catch(() => {}); // outside the room gate and not awaited: a slow presence write holds neither another send nor this receipt
|
||||
if (follow) { const at = now(); anchor(roomId, result.eventId, at).catch(error => say(`follow: send not anchored: ${error.message}`)); } // in the background too: a room-state read that hangs or fails never delays the receipt
|
||||
}
|
||||
return result;
|
||||
} catch (error) {
|
||||
say(`send failed: ${error.message}`);
|
||||
|
|
@ -887,7 +996,9 @@ export function createMouth({
|
|||
}
|
||||
}
|
||||
|
||||
return { start, syncOnce, ack, read, pendingAfter, tail, speakableRooms, send, openDirect, fetchAttachment, run, get userId() { return selfId; }, get pendingCount() { return pending.length; } };
|
||||
// A plain copy of the follow memory, for tests: anchors with their counted senders, the sender ring, and how many cap entries are held.
|
||||
const followState = () => ({ engagements: [...engagements].map(([room, e]) => ({ room, from: e.from, ends: e.ends, counted: [...e.counted] })), spoke: [...spoke].map(([room, ring]) => ({ room, senders: ring.map(r => r.sender) })), followsGiven: followsGiven.size });
|
||||
return { followState, start, syncOnce, ack, read, pendingAfter, tail, speakableRooms, send, openDirect, fetchAttachment, run, get userId() { return selfId; }, get pendingCount() { return pending.length; } };
|
||||
}
|
||||
|
||||
// The log's word for a failure: error class and HTTP status only, never the message (it can hold a URL, an mxc or a sender-chosen id).
|
||||
|
|
@ -1074,6 +1185,8 @@ export function loadRuntime(cultRoot) {
|
|||
export const replyLinkFromEnv = env => env.CHAT_PENDING_V4 === "1";
|
||||
// The fetch flag, the same way: on only for exactly "1". Off, the fetch request kind is an unknown kind.
|
||||
export const fetchFromEnv = env => env.CHAT_FETCH === "1";
|
||||
// The follow flag, the same way. Off, the mouth never emits basis "follow" (Huginn's parser refuses an unknown basis until its Basis::Follow lands).
|
||||
export const followFromEnv = env => env.CHAT_FOLLOW === "1";
|
||||
|
||||
function required(env, name) {
|
||||
if (!env[name]) throw new Error(`${name} is required`);
|
||||
|
|
@ -1127,7 +1240,7 @@ export async function main(env = process.env) {
|
|||
const mouths = new Map(personas.map(persona => {
|
||||
mkdirSync(join(stateRoot, persona), { recursive: true });
|
||||
return [persona, createMouth({
|
||||
persona, homeserver, serverName, rooms: roomsByPersona.get(persona), token: tokens.get(persona), gate, replyLink, isServed: async id => { await Promise.all([...mouths.values()].map(m => m.start().catch(() => {}))); return [...mouths.values()].some(m => m.userId === id); }, log: line => log(`${persona}: ${line}`),
|
||||
persona, homeserver, serverName, rooms: roomsByPersona.get(persona), token: tokens.get(persona), gate, replyLink, follow: followFromEnv(env), isServed: async id => { await Promise.all([...mouths.values()].map(m => m.start().catch(() => {}))); return [...mouths.values()].some(m => m.userId === id); }, log: line => log(`${persona}: ${line}`),
|
||||
images: imageRoot ? createImageStore({ root: join(imageRoot, persona), log: line => log(`${persona}: ${line}`) }) : null,
|
||||
state: cultcacheState({ runtime, path: join(stateRoot, persona, "cursor.cc") }),
|
||||
brake: createBrakeReader({ controlPath: join(controlRoot, persona, "control.cc"), readStore: runtime.readSingleFileStore, decode: runtime.msgpack.decode }),
|
||||
|
|
|
|||
Loading…
Reference in a new issue