diff options
| -rw-r--r-- | index.ts | 18 | ||||
| -rw-r--r-- | lib/nostr.ts | 255 | ||||
| -rw-r--r-- | lib/strfry.ts | 68 | ||||
| -rw-r--r-- | lib/utils.ts | 6 | ||||
| -rw-r--r-- | lib/whitelist.ts | 1 | ||||
| -rwxr-xr-x | listener.sh | 11 | ||||
| -rw-r--r-- | routes/nostr.ts | 220 | ||||
| -rw-r--r-- | routes/users.ts | 20 | ||||
| -rw-r--r-- | server.c | 124 |
9 files changed, 315 insertions, 408 deletions
@@ -6,7 +6,6 @@ import { getLocations } from "$lib/locations"; import { catchUp, check } from "$lib/payments"; import { getFx } from "$lib/rates"; import { sendHeartbeat } from "$lib/sockets"; -import { fillPool } from "$lib/nostr"; import nwc from "$lib/nwc"; import ecash from "$routes/ecash"; @@ -22,13 +21,7 @@ import rates from "$routes/rates"; import shopify from "$routes/shopify"; import users from "$routes/users"; -import sf from "$lib/strfry"; -console.log(await sf( - `scan '{"authors":["c6047f9441ed7d6d3045406e95c07cd85c778e4b8cef3ca7abac09b95c709ee5"]}'`, -)); - try { - fillPool(); getLocations(); getFx(); catchUp(); @@ -56,10 +49,7 @@ app.post("/invoice", optional, invoices.create); app.get("/nostr.json", nostr.identities); app.get("/:pubkey/followers", nostr.followers); app.get("/:pubkey/follows", nostr.follows); -app.get("/:pubkey/notes", nostr.notes); -app.get("/:pubkey/:since/messages", nostr.messages); app.get("/event/:id", nostr.event); -app.post("/event", nostr.broadcast); app.get("/info", payments.info); app.post("/payments", auth, payments.create); @@ -139,9 +129,11 @@ let port: number = parseInt(process.env["PORT"]) || 3119; app.listen({ host, port }); let logerr = (e: Error) => - e.message.includes("Invalid") || - e.message.includes("MASK") || - e.message.includes("Rate") || + // (e && + // e.message && + // (e.message.includes("Invalid") || + // e.message.includes("MASK") || + // e.message.includes("Rate"))) || console.log(e); process.on("unhandledRejection", logerr); diff --git a/lib/nostr.ts b/lib/nostr.ts index 44febae3..a1a9a245 100644 --- a/lib/nostr.ts +++ b/lib/nostr.ts @@ -1,217 +1,26 @@ import config from "$config"; +import { db } from "$lib/db"; import { l } from "$lib/logging"; -import { Relay, RelayPool, getPublicKey } from "nostr"; -import { g, s, db } from "$lib/db"; -import { fail, wait } from "$lib/utils"; -import { nip19, finalizeEvent } from "nostr-tools"; -import { emit } from "$lib/sockets"; -import { hex } from "@scure/base"; -import { Relay as ntr } from "nostr-tools/relay"; +import { fail } from "$lib/utils"; +import { nip19, finalizeEvent, getPublicKey } from "nostr-tools"; +import { Relay } from "nostr-tools/relay"; +import { SimplePool } from "nostr-tools"; -export const serverPubkey = hex.encode( - getPublicKey(nip19.decode(config.nostrKey).data), +export const serverPubkey = getPublicKey( + nip19.decode(config.nostrKey).data as Uint8Array, ); -export let coinos; -export let pool; +export let pool = new SimplePool(); -export let fillPool = () => { +export async function send(ev, url = config.nostr) { + if (!(ev && ev.id)) return; try { - pool = RelayPool(config.relays); - pool.on("open", (relay) => { - if (relay.url.includes(config.nostr)) coinos = relay; - // relay.subscribe("live", { limit: 1 }); - }); - - pool.on("event", async (_, sub, ev) => { - if (!coinos || coinos.ws.readyState !== WebSocket.OPEN) return; - if (timeouts[sub]) timeouts[sub].extend(ev); - try { - let content; - try { - content = JSON.parse(ev.content); - } catch (e) { - content = ev.content; - } - - if (sub === "live") { - let { pubkey } = ev; - - if (Math.abs(Math.floor(Date.now() / 1000) - ev.created_at) > 7200) - return; - - if (seen.includes(ev.id)) return; - seen.push(ev.id); - seen.length > 1000 && seen.shift(); - - if (coinos && ev.kind < 5) { - // coinos.send(["EVENT", ev]); - ev.user = await getUser(pubkey); - // broadcast("event", ev); - - if (ev.kind === 4) { - let uid = await g(`user:${ev.tags[0][1]}`); - if (uid) ev.recipient = await g(`user:${uid}`); - ev.author = ev.user; - emit(uid, "event", ev); - } - } - } else if (sub.includes("profile")) { - let { pubkey } = ev; - let user = await getUser(pubkey); - - if (user.updated > ev.created_at) return; - - if (content.name && user.username === pubkey.substr(0, 6)) - user.username = content.name; - - delete content.name; - - user = { - ...user, - ...content, - updated: ev.created_at, - }; - - console.log("SAVING", pubkey); - - await s(`user:${pubkey}`, user); - } else if (sub.includes("messages")) { - let pubkey = sub.split(":")[0]; - await db.sAdd(`${pubkey}:messages`, ev.id); - await s(`ev:${ev.id}`, ev); - } else if (sub.includes("follows")) { - let pubkey = sub.split(":")[0]; - await s(`${pubkey}:follows`, ev.tags); - for (let f of ev.tags) { - let [_, followPubkey] = f; - await db.sAdd(`${followPubkey}:followers`, pubkey); - } - coinos.send(ev); - } else if (sub.includes("followers")) { - let followed = sub.split(":")[0]; - let { pubkey } = ev; - await db.sAdd(`${followed}:followers`, pubkey); - coinos.send(ev); - } else if (sub.includes("notes")) { - let pubkey = sub.split(":")[0]; - await db.sAdd(pubkey, ev.id); - await s(`ev:${ev.id}`, ev); - } - } catch (e) { - console.log(e); - } - }); + let r = await Relay.connect(url); + await r.publish(ev); + r.close(); } catch (e) { - console.log("POOL", e); + console.log("nostr send error", e); } -}; - -let now = () => Math.round(Date.now() / 1000); - -let timeouts = {}; -export let q = async (sub, query, { timeout = 20000, eager = 20000 } = {}) => - new Promise(async (r, j) => { - let start = Date.now(); - let seen = []; - let rejected; - // query.since = await g(`since:${sub}`); - // if (now() - query.since < since) return r(); - - let done = { [sub]: [] }; - - let check = (s) => { - let elapsed = Date.now() - start; - if (rejected) return true; - if ( - done[s] && - (done[s].length === pool.relays.length || - (done[s].length && elapsed > eager)) - ) { - r(); - if (timeouts[s]) timeouts[s].clear(); - return true; - } - }; - - timeouts[sub] = { - async clear() { - clearTimeout(this.timer); - delete timeouts[sub]; - pool.unsubscribe(sub); - await s(`since:${sub}`, now()); - }, - extend(ev) { - if (ev) { - if (seen.includes(ev.id)) return; - seen.push(ev.id); - seen.length > 1000 && seen.shift(); - } - - clearTimeout(this.timer); - this.timer = setTimeout(() => { - j(new Error(`query timed out: ${sub} ${Date.now()}`)); - pool.unsubscribe(sub); - rejected = true; - }, timeout); - }, - timer: undefined, - }; - - timeouts[sub].extend(); - pool.subscribe(sub, query); - - pool.on("eose", (relay, s) => { - if (done[s]) done[s].push(relay.url); - check(s); - }); - - try { - await wait(() => check(sub), 100, 200); - } catch (e) { - console.log(e); - } - }); - -let seen = []; - -let getUser = async (pubkey) => { - let id = await g(`user:${pubkey}`); - let user = await g(`user:${id}`); - return ( - user || { - username: pubkey.substr(0, 6), - pubkey, - anon: true, - } - ); -}; - -function send(ev, url, opts = { timeout: 1000 }) { - let { timeout } = opts; - - return new Promise((resolve, reject) => { - let relay = Relay(url); - - function timeout_reached() { - relay.close(); - reject(new Error("Request timeout")); - } - - let timer = setTimeout(timeout_reached, timeout); - - relay.on("open", () => { - clearTimeout(timer); - timer = setTimeout(timeout_reached, timeout); - relay.send(["EVENT", ev]); - }); - - relay.on("ok", (evid, ok, msg) => { - clearTimeout(timer); - relay.close(); - resolve({ evid, ok, msg }); - }); - }); } export async function handleZap(invoice) { @@ -269,9 +78,43 @@ export async function handleZap(invoice) { await Promise.allSettled( relays.map(async (url) => { - let r = await ntr.connect(url); + let r = await Relay.connect(url); await r.publish(signed); r.close(); }), ); } + +export let getRelays = async (pubkey): Promise<any> => { + let { relays } = config; + let filter = { authors: [pubkey], kinds: [10002] }; + let event = await pool.get([config.nostr], filter); + + let read = relays; + let write = relays; + + if (event) { + + read = []; + write = []; + for (let r of event.tags) { + if (!r[2] || r[2] === "write") write.push(r[1]); + if (!r[2] || r[2] === "read") read.push(r[1]); + } + } else pool.get(relays, filter).then(send); + + return { read, write }; +}; + +export let getProfile = async (pubkey) => { + if (await db.sIsMember("noprofile", pubkey)) return; + let relays = [config.nostr]; + let filter = { authors: [pubkey], kinds: [0] }; + let ev = await pool.get(relays, filter); + if (ev) { + return JSON.parse(ev.content); + } else + getRelays(pubkey).then(({ write: relays }) => + pool.get(relays, filter).then(send), + ); +}; diff --git a/lib/strfry.ts b/lib/strfry.ts index ebc14ed3..029c8eee 100644 --- a/lib/strfry.ts +++ b/lib/strfry.ts @@ -1,30 +1,58 @@ -import { connect } from "net"; +import { createServer, connect } from "net"; +import { randomUUID } from "crypto"; -export default (cmd) => - new Promise((r, j) => { - let results = []; - let c = `/app/strfry ${cmd} 2>/dev/null\n`; - let client = connect("/sockets/ctrl", () => client.write(c)); +let exec = async (cmd) => + new Promise((resolve, reject) => { + let c = `/app/strfry ${cmd} 2>/dev/null`; + let id = randomUUID(); + const resultSocketPath = `/sockets/result_${id}`; - let s = ""; - client.on("data", (data) => { - s += data.toString(); - // console.log("S", s); + const resultServer = createServer((socket) => { + let resultBuffer = ""; - try { - let parts = s.split("}"); + let results = []; + socket.on("data", (data) => { + resultBuffer += data.toString(); + + let parts = resultBuffer.split("\n"); for (let i = 0; i < parts.length - 1; i++) { - let completeJson = parts[i] + "}"; - results.push(JSON.parse(completeJson)); + try { + let parsedObject = JSON.parse(parts[i]); + results.push(parsedObject); + } catch (e) { + console.error(parts[i]); + } } - s = parts[parts.length - 1]; - } catch (err) { - console.error("Error parsing JSON:", err.message); - } + resultBuffer = parts[parts.length - 1]; + }); + + socket.on("end", () => { + socket.write("DONE"); + resolve(results); + socket.end(); + + resultServer.close(); + }); }); - client.on("error", (e) => j(e.message)); - client.on("close", () => r(results)); + resultServer.listen(resultSocketPath, () => { + const controlClient = connect("/sockets/ctrl", () => { + const message = `${resultSocketPath} ${c}\n`; + controlClient.write(message); + controlClient.end(); + }); + + controlClient.on("error", (e) => + reject(`Control socket error: ${e.message}`), + ); + }); + + resultServer.on("error", (e) => + reject(`Result server error: ${e.message}`), + ); }); + +export let scan = (f) => exec(`scan '${JSON.stringify(f)}'\n`) +export let sync = (r, f) => exec(`sync ${r} --dir down --filter '${JSON.stringify(f)}'\n`) diff --git a/lib/utils.ts b/lib/utils.ts index a805f343..7a663c60 100644 --- a/lib/utils.ts +++ b/lib/utils.ts @@ -168,3 +168,9 @@ export function formatReceipt(items, currency) { } export let t = ({ language = "en" }) => locales[language]; + +export let chunk = (array, chunkSize) => array.reduce((all, one, i) => { + const chunkIndex = Math.floor(i / chunkSize); + (all[chunkIndex] = all[chunkIndex] || []).push(one); + return all; +}, []); diff --git a/lib/whitelist.ts b/lib/whitelist.ts index bf5b4eb9..54d3aafc 100644 --- a/lib/whitelist.ts +++ b/lib/whitelist.ts @@ -29,6 +29,7 @@ export default [ "nsec", "nwc", "payments", + "picture", "profile", "prompt", "pubkey", diff --git a/listener.sh b/listener.sh deleted file mode 100755 index 1d68bb4a..00000000 --- a/listener.sh +++ /dev/null @@ -1,11 +0,0 @@ -#!/bin/bash - -# Check if the socket file exists -if [ -S /sockets/ctrl ]; then - echo "Socket exists. Unlinking..." - rm /sockets/ctrl -fi - -# Start socat -socat UNIX-LISTEN:/sockets/ctrl,reuseaddr,fork SYSTEM:'sh -c "cat | while read line; do output=\$(eval \$line 2>&1); echo \$output; done"' - diff --git a/routes/nostr.ts b/routes/nostr.ts index d3c293c6..0fce9bdc 100644 --- a/routes/nostr.ts +++ b/routes/nostr.ts @@ -1,6 +1,15 @@ -import { bail, nada, getUser, uniq } from "$lib/utils"; -import { g, db } from "$lib/db"; -import { pool, q, serverPubkey } from "$lib/nostr"; +import config from "$config"; +import { bail, chunk, getUser, uniq } from "$lib/utils"; +import { g, s, db } from "$lib/db"; +import { send, getRelays, serverPubkey, pool } from "$lib/nostr"; + +let anon = (pubkey) => ({ + username: pubkey.substr(0, 6), + pubkey, + anon: true, + follows: [], + followers: [], +}); export default { async event(req, res) { @@ -12,91 +21,8 @@ export default { let { pubkey } = event; - event.user = (await g(`user:${pubkey}`)) || { - username: pubkey.substr(0, 6), - pubkey, - anon: true, - follows: [], - followers: [], - }; - - res.send(event); - }, - - async notes(req, res) { - let { - params: { pubkey }, - } = req; - let user = await getUser(pubkey); - - if (!user) - user = { - username: pubkey.substr(0, 6), - pubkey, - anon: true, - follows: [], - followers: [], - }; - - let params = { - kinds: [1], - authors: [pubkey], - }, - opts = { since: 0 }; - - q(`${pubkey}:notes`, params, opts).catch(nada); - - let ids = await db.sMembers(pubkey); - - let events = ids.length - ? (await db.mGet(ids.map((k) => "ev:" + k))).map(JSON.parse) - : []; - - res.send(events.map((e) => ({ ...e, user }))); - }, - - async messages(req, res) { - let { - params: { pubkey, since = 0 }, - } = req; - let params = { - kinds: [4], - authors: [pubkey], - }; - - let opts = { since }; - - await q(`${pubkey}:messages`, params, opts).catch(nada); - - params = { - kinds: [4], - "#p": [pubkey], - }; - - await q(`${pubkey}:messages`, params, opts).catch(nada); - - let messages = await db.sMembers(`${pubkey}:messages`); - - messages = await Promise.all( - messages.map(async (id) => { - let m = await g(`ev:${id}`); + event.user = (await g(`user:${pubkey}`)) || anon(pubkey); - let aid = await g(`user:${m.pubkey}`); - m.author = await g(`user:${aid}`); - - let rid = await g(`user:${m.tags[0][1]}`); - m.recipient = await g(`user:${rid}`); - - return m; - }), - ); - - res.send(messages); - }, - - async broadcast(req, res) { - let { event } = req.body; - pool.send(["EVENT", event]); res.send(event); }, @@ -106,54 +32,70 @@ export default { query: { tagsonly }, } = req; try { - let sub = `${pubkey}:follows`, - params = { - limit: 1, - kinds: [3], - authors: [pubkey], - }, - opts = { timeout: 60000, eager: 60000 }; - - q(sub, params, opts).catch(nada); + let { relays } = config; + ({ write: relays } = await getRelays(pubkey)); - let tags = (await g(`${pubkey}:follows`)) || []; - if (tagsonly) return res.send(tags); + let ev = await pool.get([config.nostr], { + kinds: [3], + authors: [pubkey], + }); + if (!ev) { + pool + .get(relays, { + kinds: [3], + authors: [pubkey], + }) + .then(send); + } + let created_at; let follows = []; + let tags = []; + if (ev) ({ created_at, tags } = ev); + if (!tags.length || tagsonly) return res.send(tags); - console.log("L", tags.length); + let cache = await g(`${pubkey}:follows`); + if (cache && cache.t >= created_at) ({ follows } = cache); + else { + let profiles = []; + let pubkeys = tags.map((t) => t[1]).filter((p) => p.length === 64); + for (let authors of chunk(pubkeys, 100)) { + let filter = { authors, kinds: [0] }; + profiles.push(...(await pool.querySync([config.nostr], filter))); + } - for (let f of tags) { - let [_, pubkey] = f; - let user = await getUser(pubkey); + for (let p of profiles) { + let { content, pubkey } = p; + let user = JSON.parse(content); - if (!(user && user.updated)) { - console.log("Q", pubkey) - q(`${pubkey}:profile:f1`, { - limit: 1, - kinds: [0], - authors: [pubkey], - }).catch(nada); - } + let cache = await g(`${pubkey}:followers`); + let followers; + if (cache && cache.t >= created_at) ({ followers } = cache); + else { + let filter = { "#p": [pubkey], kinds: [3] }; + followers = await pool.querySync([config.nostr], filter); + s(`${pubkey}:followers`, { followers, t: created_at }); + } - if (!user) { - user = { - username: pubkey.substr(0, 6), - pubkey, - anon: true, - }; + user.followers = followers.length; + user.pubkey = pubkey; + follows.push(user); } - follows.push(user); - } + let followKeys = follows.map((f) => f.pubkey); + let missing = pubkeys + .filter((p) => !followKeys.includes(p)) + .map((pubkey) => anon(pubkey)); - follows = uniq(follows, (e) => e.pubkey); - follows.sort( - (a, b) => a.username && a.username.localeCompare(b.username), - ); + follows.sort((a: any, b: any) => b.followers - a.followers); + follows.push(...missing); + + s(`${pubkey}:follows`, { follows, t: created_at }); + } res.send(follows); } catch (e) { + console.log("follows fail", e); bail(res, e.message); } }, @@ -163,38 +105,10 @@ export default { params: { pubkey }, } = req; try { - let pubkeys = [ - ...new Set([...(await db.sMembers(`${pubkey}:followers`))]), - ]; - - let followers = []; - - q( - `${pubkey}:followers`, - { kinds: [3], "#p": [pubkey] }, - { timeout: 60000, eager: 60000 }, - ).catch(nada); + let pubkeys: any = await q({ "#p": [pubkey] }); + pubkeys = pubkeys.map((e) => e.pubkey); - for (let pubkey of pubkeys) { - let user = await getUser(pubkey); - - if (!user?.updated) { - q(`${pubkey}:profile:f2`, { - limit: 1, - kinds: [0], - authors: [pubkey], - }).catch(nada); - } - - if (!user) - user = { - username: pubkey.substr(0, 6), - pubkey, - anon: true, - }; - - followers.push(user); - } + let followers: any = await q({ authors: [pubkeys], kinds: [0] }); followers = uniq(followers, (e) => e.pubkey); followers.sort( diff --git a/routes/users.ts b/routes/users.ts index 18ef7e3b..f907340e 100644 --- a/routes/users.ts +++ b/routes/users.ts @@ -15,7 +15,7 @@ import { mail, templates } from "$lib/mail"; import upload from "$lib/upload"; import rpc from "@coinos/rpc"; import { nip19 } from "nostr-tools"; -import { hex } from "@scure/base"; +import { getProfile } from "$lib/nostr"; export default { upload, @@ -78,19 +78,27 @@ export default { try { if (key.startsWith("npub")) { try { - key = hex.encode(nip19.decode(key).data as Uint8Array); + key = nip19.decode(key).data; } catch (e) {} } let user = await getUser(key); - if (!user && key.length === 64) { + if (key.length === 64) { + let nostr: any = await getProfile(key); + if (nostr) { + nostr.username = nostr.name; + nostr.display = nostr.display_name || nostr.displayName; + } + + if (user) user.anon = false; + user = { + ...nostr, currency: "USD", - username: key, - display: key.substr(0, 6), pubkey: key, anon: true, + ...user, }; } @@ -105,8 +113,10 @@ export default { "display", "id", "hidepay", + "lud16", "memoPrompt", "npub", + "picture", "profile", "prompt", "pubkey", diff --git a/server.c b/server.c new file mode 100644 index 00000000..cd49f18b --- /dev/null +++ b/server.c @@ -0,0 +1,124 @@ +#include <stdio.h> +#include <stdlib.h> +#include <string.h> +#include <sys/socket.h> +#include <sys/un.h> +#include <unistd.h> + +#define SOCKET_PATH "/sockets/ctrl" +#define BUFFER_SIZE 1024 + +void handle_client(const char *result_socket, const char *command); + +int main() { + int server_fd, client_fd; + struct sockaddr_un server_addr; + char buffer[BUFFER_SIZE]; + + unlink(SOCKET_PATH); + + server_fd = socket(AF_UNIX, SOCK_STREAM, 0); + if (server_fd == -1) { + perror("socket failed"); + exit(EXIT_FAILURE); + } + + memset(&server_addr, 0, sizeof(struct sockaddr_un)); + server_addr.sun_family = AF_UNIX; + strncpy(server_addr.sun_path, SOCKET_PATH, sizeof(server_addr.sun_path) - 1); + + if (bind(server_fd, (struct sockaddr *)&server_addr, + sizeof(struct sockaddr_un)) == -1) { + perror("bind failed"); + close(server_fd); + exit(EXIT_FAILURE); + } + + if (listen(server_fd, 5) == -1) { + perror("listen failed"); + close(server_fd); + exit(EXIT_FAILURE); + } + + printf("Server listening on %s\n", SOCKET_PATH); + + while (1) { + + client_fd = accept(server_fd, NULL, NULL); + if (client_fd == -1) { + perror("accept failed"); + continue; + } + + memset(buffer, 0, BUFFER_SIZE); + ssize_t num_bytes = read(client_fd, buffer, BUFFER_SIZE - 1); + if (num_bytes > 0) { + + buffer[num_bytes] = '\0'; + + printf("Received: %s\n", buffer); + + char result_socket[BUFFER_SIZE], command[BUFFER_SIZE]; + if (sscanf(buffer, "%s %[^\n]", result_socket, command) == 2) { + printf("Result socket: %s\n", result_socket); + printf("Command: %s\n", command); + + handle_client(result_socket, command); + } else { + fprintf(stderr, "Invalid input format\n"); + } + } + + close(client_fd); + } + + close(server_fd); + unlink(SOCKET_PATH); + return 0; +} + +void handle_client(const char *result_socket, const char *command) { + pid_t pid = fork(); + + if (pid == 0) { + + int result_fd = socket(AF_UNIX, SOCK_STREAM, 0); + if (result_fd == -1) { + perror("result socket failed"); + exit(EXIT_FAILURE); + } + + struct sockaddr_un result_addr; + memset(&result_addr, 0, sizeof(struct sockaddr_un)); + result_addr.sun_family = AF_UNIX; + strncpy(result_addr.sun_path, result_socket, + sizeof(result_addr.sun_path) - 1); + + if (connect(result_fd, (struct sockaddr *)&result_addr, + sizeof(struct sockaddr_un)) == -1) { + perror("connect to result socket failed"); + close(result_fd); + exit(EXIT_FAILURE); + } + + FILE *cmd_output = popen(command, "r"); + if (cmd_output == NULL) { + perror("popen failed"); + close(result_fd); + exit(EXIT_FAILURE); + } + + char output_buffer[BUFFER_SIZE]; + while (fgets(output_buffer, sizeof(output_buffer), cmd_output) != NULL) { + write(result_fd, output_buffer, strlen(output_buffer)); + } + + pclose(cmd_output); + shutdown(result_fd, SHUT_WR); + close(result_fd); + unlink(result_socket); + exit(0); + } else if (pid < 0) { + perror("fork failed"); + } +} |