- commit
- eb8640f806aa028930799270c737364cf64ce166
- parent
- 50319588895932a761dc05d0f18bcc9ee065a81c
- author
- Dami <dami.etoile@gmail.com>
| M | src/ | 6 | +- |
| M | src/ | 124 | +++++------------------------- |
| M | src/ | 94 | ++++++++++++++++++++++- |
@@ -14,7 +14,7 @@ import { badBlockedWord, blockedWords, gateFields, isWriteRule, limitFields, vie
1414 import { isReplaceable, isProtected, publicFields, dumpFields, validIP } from "./settings.ts";1515 export { validIP };1616 import { checkPullURL } from "./pull.ts";17-import { VIEWS, viewsSummary } from "./views.ts";17+import { VIEWS, publishView, viewsSummary } from "./views.ts";1818 import { MAX_PINS } from "./groups.ts";1919 import { relaysFromList } from "./jobs.ts";2020 import { notify, notifySettings } from "./notify.ts";@@ -183,7 +183,7 @@ export const METHODS: Record<string, Method> = {
183183 // the read rule and the directory decide each one's audience.184184 const names = patch.views && typeof patch.views === "object" ? Object.keys(patch.views as object) : [];185185 if (clean.reads !== undefined || clean.directoryPublic !== undefined) for (const v of VIEWS) if (!names.includes(v.name)) names.push(v.name);186- for (const name of names) await relay.publishView(name);186+ for (const name of names) await publishView(relay, name);187187 return reply({ result: s.policy });188188 },189189 },@@ -512,7 +512,7 @@ export const METHODS: Record<string, Method> = {
512512 }513513 await relay.publishMembership();514514 // The read rule and the directory moved, so every view's audience may have.515- for (const v of VIEWS) await relay.publishView(v.name);515+ for (const v of VIEWS) await publishView(relay, v.name);516516 return reply({ result: { ...s.policy, job } });517517 },518518 },@@ -38,9 +38,9 @@ import { SIGNER_JS } from "./gen/signer.ts";
3838 import { isPagePath, pages } from "./pages.ts";3939 import { notify, fuelLow, fuelText } from "./notify.ts";4040 import { card } from "./card.ts";41-import { PRESENCE_THROTTLE_S, VIEWS, latestStored, nip11Views, presenceTags, serveView, viewByName, viewD, viewEvent, viewOn, viewStored, type ViewRun } from "./views.ts";41+import { PRESENCE_THROTTLE_S, markView, nip11Views, notePresence, serveView, viewsTick } from "./views.ts";4242 import type { Blob } from "./blossom.ts";43-import { KIND_VANISH, KIND_AUTH, KIND_REPORT, KIND_NOSTR_CONNECT, KIND_PRESENCE, KIND_GROUP_MEMBERS, KIND_GROUP_PINS, KIND_RELAY_DISCOVERY } from "./kinds.ts";43+import { KIND_VANISH, KIND_AUTH, KIND_REPORT, KIND_NOSTR_CONNECT, KIND_GROUP_MEMBERS, KIND_GROUP_PINS, KIND_RELAY_DISCOVERY } from "./kinds.ts";4444 4545 export interface Env {4646 RELAY: DurableObjectNamespace<Relay>;@@ -145,11 +145,11 @@ export class Relay extends DurableObject<Env> {
145145 // Presence (views.ts): who wrote lately, in memory; the sockets are asked146146 // directly. A broadcast is at most one per PRESENCE_THROTTLE_S.147147 presenceActive = new Map<string, number>();148- private presenceAt = 0;149- private presenceTimer: ReturnType<typeof setTimeout> | null = null;148+ presenceAt = 0;149+ presenceTimer: ReturnType<typeof setTimeout> | null = null;150150 // Write-triggered views wait a moment so a burst republishes once.151- private viewDirty = new Set<string>();152- private viewTimer: ReturnType<typeof setTimeout> | null = null;151+ viewDirty = new Set<string>();152+ viewTimer: ReturnType<typeof setTimeout> | null = null;153153 // The custom hostnames client, null when the host has not enabled them.154154 hostnames: Hostnames | null;155155 @@ -250,7 +250,7 @@ export class Relay extends DurableObject<Env> {
250250 events.push(this.identity.roster(this.settings.members(), t));251251 events.push(...this.identity.group(f, t));252252 this.emit(events, t);253- this.markView("profiles");253+ markView(this, "profiles");254254 await this.publishDiscovery();255255 }256256 @@ -274,6 +274,13 @@ export class Relay extends DurableObject<Env> {
274274 275275 // ---- views (views.ts) ----276276 277+ // rowsWritten is the running count of rows written since the last flush,278+ // for a view to say what its fold cost.279+ rowsWritten(): number {280+ this.tally();281+ return this.meter.rowsWritten;282+ }283+277284 // authedNow lists every pubkey with an authenticated socket open.278285 authedNow(): string[] {279286 const out = new Set<string>();@@ -281,96 +288,6 @@ export class Relay extends DurableObject<Env> {
281288 return [...out];282289 }283290 284- async viewRuns(name: string): Promise<ViewRun[]> {285- return (await this.ctx.storage.get<ViewRun[]>("view-runs:" + name)) ?? [];286- }287-288- // publishView folds one stored view and writes its record, or takes the289- // record down when the view is off, members-only or empty. The rows the290- // fold wrote are recorded so the console and the digest can say what a291- // view costs.292- async publishView(name: string): Promise<void> {293- const v = viewByName(name);294- if (!v || v.trigger === "live" || this.settings.policy.owner === "") return;295- await this.identity.ensure();296- const self = this.identity.pubkey;297- this.tally();298- const before = this.meter.rowsWritten;299- const takeDown = () => {300- for (const r of this.sql.exec<{ id: string }>(`SELECT e.id FROM events e JOIN tags t ON t.event_id=e.id WHERE e.kind=30078 AND e.pubkey=? AND t.name='d' AND t.value=?`, self, viewD(name)).toArray()) this.store.deleteEvent(r.id);301- };302- if (!viewOn(this, name) || !viewStored(this, v)) takeDown();303- else {304- const fold = v.fold(this);305- if (!fold) takeDown();306- else this.emit([viewEvent(this, v, fold, true)], now());307- }308- this.tally();309- const runs = (await this.viewRuns(name)).slice(-59);310- runs.push({ at: now(), rows: Math.max(0, this.meter.rowsWritten - before) });311- await this.ctx.storage.put("view-runs:" + name, runs);312- if (v.fingerprint) await this.ctx.storage.put("view-fp:" + name, v.fingerprint(this));313- }314-315- // markView republishes a write-triggered view shortly, once for a burst.316- markView(name: string) {317- if (!viewOn(this, name) || this.settings.policy.owner === "") return;318- this.viewDirty.add(name);319- if (this.viewTimer) return;320- this.viewTimer = setTimeout(() => {321- this.viewTimer = null;322- const names = [...this.viewDirty];323- this.viewDirty.clear();324- void (async () => {325- for (const n of names) {326- try {327- await this.publishView(n);328- } catch (err) {329- console.log("view " + n + " failed: " + (err instanceof Error ? err.message : String(err)));330- }331- }332- })();333- }, 10_000);334- }335-336- // viewsTick runs the scheduled views: daily ones once a day, hourly ones337- // once an hour and only when their fingerprint moved or nothing is stored.338- private async viewsTick(t: number) {339- if (this.settings.policy.owner === "") return;340- for (const v of VIEWS) {341- if (v.trigger === "live" || !viewOn(this, v.name)) continue;342- const last = (await this.ctx.storage.get<number>("view-at:" + v.name)) ?? 0;343- const period = v.trigger === "hourly" ? 3600 : 86400;344- if (t - last < period - 300) continue;345- await this.ctx.storage.put("view-at:" + v.name, t);346- if (v.fingerprint && viewStored(this, v) && latestStored(this, v.name) !== null) {347- const fp = v.fingerprint(this);348- if (fp === (await this.ctx.storage.get<string>("view-fp:" + v.name))) continue;349- }350- await this.publishView(v.name);351- }352- }353-354- // notePresence records a writer and broadcasts the presence view when the355- // throttle allows, otherwise once it does.356- notePresence(pubkey?: string) {357- if (pubkey) this.presenceActive.set(pubkey, now());358- if (!this.identity.pubkey || !viewOn(this, "presence") || this.settings.policy.owner === "") return;359- const wait = this.presenceAt + PRESENCE_THROTTLE_S - Date.now() / 1000;360- if (wait <= 0) this.broadcastPresence();361- else if (!this.presenceTimer) {362- this.presenceTimer = setTimeout(() => {363- this.presenceTimer = null;364- this.broadcastPresence();365- }, wait * 1000);366- }367- }368-369- private broadcastPresence() {370- this.presenceAt = Date.now() / 1000;371- this.broadcast(this.identity.sign(KIND_PRESENCE, [["d", viewD("presence")], ...presenceTags(this)], ""));372- }373-374291 // publishPins signs the group's pin list (39005); an empty list takes the375292 // record down.376293 async publishPins() {@@ -389,7 +306,8 @@ export class Relay extends DurableObject<Env> {
389306 return r && r !== "member" ? [r] : [];390307 }391308 392- private emit(events: Event[], t: number) {309+ // emit stores relay-signed events and broadcasts the ones that were new.310+ emit(events: Event[], t: number) {393311 for (const e of events) {394312 const err = this.store.save(e, t);395313 if (!err) this.broadcast(e);@@ -1101,7 +1019,7 @@ export class Relay extends DurableObject<Env> {
11011019 this.syncs.delete(ws);11021020 this.buckets.delete(ws);11031021 this.flushUsage();1104- this.notePresence();1022+ notePresence(this);11051023 }11061024 11071025 async webSocketError(ws: WebSocket) {@@ -1226,8 +1144,8 @@ export class Relay extends DurableObject<Env> {
12261144 else this.ensureAlarm();12271145 if (e.kind === 0 && conn) claimFromProfile(this.settings, e.content, e.pubkey, conn.host, t);12281146 if (conn) void this.succession.seen(e.pubkey);1229- if (conn) this.notePresence(e.pubkey);1230- if (e.kind === 30023) this.markView("articles");1147+ if (conn) notePresence(this, e.pubkey);1148+ if (e.kind === 30023) markView(this, "articles");12311149 return { ok: true, msg: "", stored: true };12321150 }12331151 @@ -1253,7 +1171,7 @@ export class Relay extends DurableObject<Env> {
12531171 this.fuel.chargeStorage(t, this.eventBytes(), this.mediaBytes());12541172 this.store.drain();12551173 this.sweepRetention(t);1256- await this.viewsTick(t);1174+ await viewsTick(this, t);12571175 await this.publishDiscovery();12581176 // Fuel notice: once when it turns low, then once a day while it stays low.12591177 if (this.settings.policy.notify.fuel) {@@ -1415,7 +1333,7 @@ export class Relay extends DurableObject<Env> {
14151333 if (hostOf(tagValues(e, "relay")[0] ?? "") !== hostOf("ws://" + s.host)) return fail("invalid: auth relay tag does not name this relay");14161334 if (!s.authed.includes(e.pubkey)) s.authed.push(e.pubkey);14171335 void this.succession.seen(e.pubkey);1418- this.notePresence();1336+ notePresence(this);14191337 this.persist(ws, s);14201338 this.send(ws, "OK", e.id, true, "");14211339 }@@ -319,7 +319,7 @@ export async function serveView(relay: Relay, req: Request, verify: (header: str
319319 export async function viewsSummary(relay: Relay) {320320 const out = [];321321 for (const v of VIEWS) {322- const runs = await relay.viewRuns(v.name);322+ const runs = await viewRuns(relay, v.name);323323 const last = runs[runs.length - 1] ?? null;324324 out.push({ name: v.name, about: v.about, trigger: v.trigger, audience: v.audience(relay), on: viewOn(relay, v.name), stored: viewStored(relay, v), last, path: "/view/" + v.name });325325 }@@ -329,6 +329,96 @@ export async function viewsSummary(relay: Relay) {
329329 // viewRowsSince sums the rows the views wrote in a window, for the digest.330330 export async function viewRowsSince(relay: Relay, since: number): Promise<number> {331331 let rows = 0;332- for (const v of VIEWS) for (const r of await relay.viewRuns(v.name)) if (r.at >= since) rows += r.rows;332+ for (const v of VIEWS) for (const r of await viewRuns(relay, v.name)) if (r.at >= since) rows += r.rows;333333 return rows;334334 }335+336+// ---- publishing: the relay's side, called from the socket handlers and the alarm ----337+338+export async function viewRuns(relay: Relay, name: string): Promise<ViewRun[]> {339+ return (await relay.storage.get<ViewRun[]>("view-runs:" + name)) ?? [];340+}341+342+// publishView folds one stored view and writes its record, or takes the343+// record down when the view is off, members-only or empty. The rows the344+// fold wrote are recorded so the console and the digest can say what a345+// view costs.346+export async function publishView(relay: Relay, name: string): Promise<void> {347+ const v = viewByName(name);348+ if (!v || v.trigger === "live" || relay.settings.policy.owner === "") return;349+ await relay.identity.ensure();350+ const self = relay.identity.pubkey;351+ const before = relay.rowsWritten();352+ const takeDown = () => {353+ for (const r of relay.sql.exec<{ id: string }>(`SELECT e.id FROM events e JOIN tags t ON t.event_id=e.id WHERE e.kind=30078 AND e.pubkey=? AND t.name='d' AND t.value=?`, self, viewD(name)).toArray()) relay.store.deleteEvent(r.id);354+ };355+ if (!viewOn(relay, name) || !viewStored(relay, v)) takeDown();356+ else {357+ const fold = v.fold(relay);358+ if (!fold) takeDown();359+ else relay.emit([viewEvent(relay, v, fold, true)], now());360+ }361+ const runs = (await viewRuns(relay, name)).slice(-59);362+ runs.push({ at: now(), rows: Math.max(0, relay.rowsWritten() - before) });363+ await relay.storage.put("view-runs:" + name, runs);364+ if (v.fingerprint) await relay.storage.put("view-fp:" + name, v.fingerprint(relay));365+}366+367+// markView republishes a write-triggered view shortly, once for a burst.368+export function markView(relay: Relay, name: string) {369+ if (!viewOn(relay, name) || relay.settings.policy.owner === "") return;370+ relay.viewDirty.add(name);371+ if (relay.viewTimer) return;372+ relay.viewTimer = setTimeout(() => {373+ relay.viewTimer = null;374+ const names = [...relay.viewDirty];375+ relay.viewDirty.clear();376+ void (async () => {377+ for (const n of names) {378+ try {379+ await publishView(relay, n);380+ } catch (err) {381+ console.log("view " + n + " failed: " + (err instanceof Error ? err.message : String(err)));382+ }383+ }384+ })();385+ }, 10_000);386+}387+388+// viewsTick runs the scheduled views: daily ones once a day, hourly ones389+// once an hour and only when their fingerprint moved or nothing is stored.390+export async function viewsTick(relay: Relay, t: number) {391+ if (relay.settings.policy.owner === "") return;392+ for (const v of VIEWS) {393+ if (v.trigger === "live" || !viewOn(relay, v.name)) continue;394+ const last = (await relay.storage.get<number>("view-at:" + v.name)) ?? 0;395+ const period = v.trigger === "hourly" ? 3600 : 86400;396+ if (t - last < period - 300) continue;397+ await relay.storage.put("view-at:" + v.name, t);398+ if (v.fingerprint && viewStored(relay, v) && latestStored(relay, v.name) !== null) {399+ const fp = v.fingerprint(relay);400+ if (fp === (await relay.storage.get<string>("view-fp:" + v.name))) continue;401+ }402+ await publishView(relay, v.name);403+ }404+}405+406+// notePresence records a writer and broadcasts the presence view when the407+// throttle allows, otherwise once it does.408+export function notePresence(relay: Relay, pubkey?: string) {409+ if (pubkey) relay.presenceActive.set(pubkey, now());410+ if (!relay.identity.pubkey || !viewOn(relay, "presence") || relay.settings.policy.owner === "") return;411+ const wait = relay.presenceAt + PRESENCE_THROTTLE_S - Date.now() / 1000;412+ if (wait <= 0) broadcastPresence(relay);413+ else if (!relay.presenceTimer) {414+ relay.presenceTimer = setTimeout(() => {415+ relay.presenceTimer = null;416+ broadcastPresence(relay);417+ }, wait * 1000);418+ }419+}420+421+function broadcastPresence(relay: Relay) {422+ relay.presenceAt = Date.now() / 1000;423+ relay.broadcast(relay.identity.sign(KIND_PRESENCE, [["d", viewD("presence")], ...presenceTags(relay)], ""));424+}