Kieran/snort
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor: live streams

Browse Source
This commit is contained in:
kieran
2024-09-12 19:46:04 +01:00
parent 9049f337b0
commit c74af0159c
9 changed files with 99 additions and 123 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
packages/worker-relay/src/interface.ts
View File

@ -64,7 +64,6 @@ export class WorkerRelayInterface {
}
async delete(req: ReqCommand) {
console.debug("DELETE", req);
return await this.#workerRpc<ReqCommand, Array<string>>("delete", req);
}

3
packages/worker-relay/src/queue.ts
View File

@ -4,7 +4,7 @@ export interface WorkQueueItem {
reject(e: unknown): void;
}
export async function processWorkQueue(queue?: Array<WorkQueueItem>, queueDelay = 200) {
export async function processWorkQueue(queue?: Array<WorkQueueItem>) {
while (queue && queue.length > 0) {
const v = queue.shift();
if (v) {
@ -16,7 +16,6 @@ export async function processWorkQueue(queue?: Array<WorkQueueItem>, queueDelay
}
}
}
setTimeout(() => processWorkQueue(queue, queueDelay), queueDelay);
}
export const barrierQueue = async <T>(queue: Array<WorkQueueItem>, then: () => Promise<T>): Promise<T> => {

BIN
packages/worker-relay/src/sqlite/sqlite3.wasm
View File

Binary file not shown.

150
packages/worker-relay/src/worker.ts
View File

@ -24,22 +24,20 @@ let eventWriteQueue: Array<NostrEvent> = [];
async function insertBatch() {
// Only insert event batches when the command queue is empty
// This is to make req's execute first and not block them
if (eventWriteQueue.length > 0 && cmdQueue.length === 0) {
await barrierQueue(cmdQueue, async () => {
const start = unixNowMs();
const timeLimit = 1000;
if (relay) {
while (eventWriteQueue.length > 0) {
if (unixNowMs() - start >= timeLimit) {
//console.debug("Yield insert, queue length: ", eventWriteQueue.length, ", cmds: ", cmdQueue.length);
break;
}
const batch = eventWriteQueue.splice(0, insertBatchSize);
eventWriteQueue = eventWriteQueue.slice(batch.length);
relay.eventBatch(batch);
if (eventWriteQueue.length > 0) {
const start = unixNowMs();
const timeLimit = 1000;
if (relay) {
while (eventWriteQueue.length > 0) {
if (unixNowMs() - start >= timeLimit) {
//console.debug("Yield insert, queue length: ", eventWriteQueue.length, ", cmds: ", cmdQueue.length);
break;
}
const batch = eventWriteQueue.splice(0, insertBatchSize);
eventWriteQueue = eventWriteQueue.slice(batch.length);
relay.eventBatch(batch);
}
});
}
}
setTimeout(() => insertBatch(), 100);
}
@ -47,7 +45,6 @@ async function insertBatch() {
const cmdQueue: Array<WorkQueueItem> = [];
try {
setTimeout(() => insertBatch(), 100);
processWorkQueue(cmdQueue, 50);
} catch (e) {
console.error(e);
}
@ -75,23 +72,21 @@ const handleMsg = async (port: MessagePort | DedicatedWorkerGlobalScope, ev: Mes
break;
}
case "init": {
await barrierQueue(cmdQueue, async () => {
const args = msg.args as InitAargs;
insertBatchSize = args.insertBatchSize ?? 10;
try {
if ("WebAssembly" in self) {
relay = new SqliteRelay();
} else {
relay = new InMemoryRelay();
}
await relay.init(args.databasePath);
} catch (e) {
console.error("Fallback to InMemoryRelay", e);
const args = msg.args as InitAargs;
insertBatchSize = args.insertBatchSize ?? 10;
try {
if ("WebAssembly" in self) {
relay = new SqliteRelay();
} else {
relay = new InMemoryRelay();
await relay.init(args.databasePath);
}
reply(msg.id, true);
});
await relay.init(args.databasePath);
} catch (e) {
console.error("Fallback to InMemoryRelay", e);
relay = new InMemoryRelay();
await relay.init(args.databasePath);
}
reply(msg.id, true);
break;
}
case "event": {
@ -105,85 +100,68 @@ const handleMsg = async (port: MessagePort | DedicatedWorkerGlobalScope, ev: Mes
break;
}
case "close": {
await barrierQueue(cmdQueue, async () => {
const res = relay!.close();
reply(msg.id, res);
});
const res = relay!.close();
reply(msg.id, res);
break;
}
case "req": {
await barrierQueue(cmdQueue, async () => {
const req = msg.args as ReqCommand;
const filters = req.slice(2) as Array<ReqFilter>;
const results: Array<string | NostrEvent> = [];
const ids = new Set<string>();
for (const r of filters) {
const rx = relay!.req(req[1], r);
for (const x of rx) {
if ((typeof x === "string" && ids.has(x)) || ids.has((x as NostrEvent).id)) {
continue;
}
ids.add(typeof x === "string" ? x : (x as NostrEvent).id);
results.push(x);
const req = msg.args as ReqCommand;
const filters = req.slice(2) as Array<ReqFilter>;
const results: Array<string | NostrEvent> = [];
const ids = new Set<string>();
for (const r of filters) {
const rx = relay!.req(req[1], r);
for (const x of rx) {
if ((typeof x === "string" && ids.has(x)) || ids.has((x as NostrEvent).id)) {
continue;
}
ids.add(typeof x === "string" ? x : (x as NostrEvent).id);
results.push(x);
}
reply(msg.id, results);
});
}
reply(msg.id, results);
break;
}
case "count": {
await barrierQueue(cmdQueue, async () => {
const req = msg.args as ReqCommand;
let results = 0;
const filters = req.slice(2) as Array<ReqFilter>;
for (const r of filters) {
const c = relay!.count(r);
results += c;
}
reply(msg.id, results);
});
const req = msg.args as ReqCommand;
let results = 0;
const filters = req.slice(2) as Array<ReqFilter>;
for (const r of filters) {
const c = relay!.count(r);
results += c;
}
reply(msg.id, results);
break;
}
case "delete": {
console.debug("DELETE", msg.args);
await barrierQueue(cmdQueue, async () => {
const req = msg.args as ReqCommand;
let results = [];
const filters = req.slice(2) as Array<ReqFilter>;
for (const r of filters) {
const c = relay!.delete(r);
results.push(...c);
}
reply(msg.id, results);
});
const req = msg.args as ReqCommand;
let results = [];
const filters = req.slice(2) as Array<ReqFilter>;
for (const r of filters) {
const c = relay!.delete(r);
results.push(...c);
}
reply(msg.id, results);
break;
}
case "summary": {
await barrierQueue(cmdQueue, async () => {
const res = relay!.summary();
reply(msg.id, res);
});
const res = relay!.summary();
reply(msg.id, res);
break;
}
case "dumpDb": {
await barrierQueue(cmdQueue, async () => {
const res = await relay!.dump();
reply(msg.id, res);
});
const res = await relay!.dump();
reply(msg.id, res);
break;
}
case "forYouFeed": {
await barrierQueue(cmdQueue, async () => {
const res = await getForYouFeed(relay!, msg.args as string);
reply(msg.id, res);
});
const res = await getForYouFeed(relay!, msg.args as string);
reply(msg.id, res);
break;
}
case "setEventMetadata": {
await barrierQueue(cmdQueue, async () => {
const [id, metadata] = msg.args as [string, EventMetadata];
relay!.setEventMetadata(id, metadata);
});
const [id, metadata] = msg.args as [string, EventMetadata];
relay!.setEventMetadata(id, metadata);
break;
}
default: {