coracle/welshman
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Re-work publish/thunk status

Browse Source
This commit is contained in:
Jon Staab
2025-10-01 16:59:16 -07:00
parent a2d519766d
commit d01f844078
21 changed files with 225 additions and 190 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/app/publishing-events.md
+5 -5
View File
@@ -32,10 +32,10 @@ const publish = async (content: string) => {
delay: 3000, // 3s window for abort
})
// Track publish status
thunk.status.subscribe(statuses => {
for (const [url, {status, message}] of Object.entries(statuses)) {
console.log(`${url}: ${status} ${message}`)
// Track publish results
thunk.subscribe($thunk => {
for (const [url, result] of Object.entries($thunk.results)) {
console.log(`${url}: ${result.status} - ${result.detail}`)
}
})
@@ -47,7 +47,7 @@ const publish = async (content: string) => {
}, 1000)
// Wait for completion
await thunk.result
await thunk.complete
}
```
docs/net/publish.md
+24 -7
View File
@@ -14,11 +14,24 @@ Status values for publish operations:
- `Timeout` - Request timed out
- `Aborted` - Request was aborted
## Types
### PublishResult
Result object for publish operations:
- `relay` - The relay URL
- `status` - PublishStatus enum value
- `detail` - Human-readable status message
### PublishResultsByRelay
Type alias for `Record<string, PublishResult>` - maps relay URLs to their publish results.
## Functions
### publishOne(options)
Publishes an event to a single relay and returns a promise that resolves with the publish status.
Publishes an event to a single relay and returns a promise that resolves with a `PublishResult`.
**Options:**
- `event` - The signed event to publish
@@ -26,11 +39,11 @@ Publishes an event to a single relay and returns a promise that resolves with th
- `signal?` - AbortSignal for cancellation
- `timeout?` - Timeout in milliseconds (default: 10000)
- `context?` - Adapter context
- Callback functions: `onSuccess`, `onFailure`, `onPending`, `onTimeout`, `onAborted`, `onComplete`
- Callback functions (all receive `PublishResult`): `onSuccess`, `onFailure`, `onPending`, `onTimeout`, `onAborted`, `onComplete`
### publish(options)
Publishes an event to multiple relays in parallel and returns a status object mapping relay URLs to their publish status.
Publishes an event to multiple relays in parallel and returns a `PublishResultsByRelay` object mapping relay URLs to their publish results.
## Example
@@ -41,13 +54,17 @@ const event = {
// ... signed event
}
const statusByRelay = await publish({
const resultsByRelay = await publish({
event,
relays: ["wss://relay1.com", "wss://relay2.com"],
timeout: 5000,
onSuccess: (detail, relay) => console.log(`Published to ${relay}`),
onFailure: (detail, relay) => console.log(`Failed on ${relay}: ${detail}`)
onSuccess: (result) => console.log(`Published to ${result.relay}: ${result.detail}`),
onFailure: (result) => console.log(`Failed on ${result.relay}: ${result.detail}`)
})
console.log(statusByRelay) // { "wss://relay1.com": "success", "wss://relay2.com": "failure" }
console.log(resultsByRelay)
// {
// "wss://relay1.com": {relay: "wss://relay1.com", status: PublishStatus.Success, detail: "ok"},
// "wss://relay2.com": {relay: "wss://relay2.com", status: PublishStatus.Failure, detail: "invalid: ..."}
// }
```
package.json
+1 -1
View File
@@ -1,7 +1,7 @@
{
"name": "@welshman",
"private": true,
"version": "0.5.0",
"version": "0.5.1",
"workspaces": [
"packages/*"
],
packages/app/__tests__/thunk.test.ts
+3 -3
View File
@@ -104,14 +104,14 @@ describe("thunk", () => {
// Wait for initial async operations
await vi.runAllTimersAsync()
expect(thunk.status[LOCAL_RELAY_URL]).toEqual(PublishStatus.Success)
expect(thunk.results[LOCAL_RELAY_URL].status).toEqual(PublishStatus.Success)
// Verify tracker was called on success
expect(track).toHaveBeenCalledWith(thunk.event.id, LOCAL_RELAY_URL)
await vi.runAllTimersAsync()
await thunk.complete
const finalStatus = await thunk.result
expect(finalStatus).toEqual({[LOCAL_RELAY_URL]: PublishStatus.Success})
expect(thunk.results[LOCAL_RELAY_URL].status).toEqual(PublishStatus.Success)
})
})
packages/app/package.json
+1 -1
View File
@@ -1,6 +1,6 @@
{
"name": "@welshman/app",
"version": "0.5.0",
"version": "0.5.1",
"author": "hodlbod",
"license": "MIT",
"description": "A collection of svelte stores for use in building nostr client applications.",
packages/app/src/relays.ts
+22 -9
View File
@@ -85,17 +85,30 @@ export const relaysByPubkey = derived(relays, $relays =>
)
export const fetchRelayProfiles = async (urls: string[]) => {
const base = appContext.dufflepudUrl
if (!base) {
throw new Error("ctx.app.dufflepudUrl is required to fetch relay metadata")
}
const res: any = await postJson(`${base}/relay/info`, {urls})
const profilesByUrl = new Map<string, RelayProfile>()
for (const {url, info} of res?.data || []) {
profilesByUrl.set(url, info)
if (appContext.dufflepudUrl) {
const res: any = await postJson(`${appContext.dufflepudUrl}/relay/info`, {urls})
for (const {url, info} of res?.data || []) {
profilesByUrl.set(url, info)
}
} else {
await Promise.all(
urls.map(async url => {
try {
const res = await fetch(url.replace(/^ws/, "http"), {
headers: {
Accept: "application/nostr+json",
},
})
profilesByUrl.set(url, await res.json())
} catch (e) {
// pass
}
}),
)
}
return profilesByUrl
packages/app/src/thunk.ts
+67 -64
View File
@@ -26,7 +26,13 @@ import {
isUnwrappedEvent,
isSignedEvent,
} from "@welshman/util"
import {publish, PublishStatus, PublishOptions, PublishStatusByRelay} from "@welshman/net"
import {
publish,
PublishStatus,
PublishResult,
PublishOptions,
PublishResultsByRelay,
} from "@welshman/net"
import {repository, tracker} from "./core.js"
import {pubkey, getSession, getSigner} from "./session.js"
@@ -57,21 +63,28 @@ export class Thunk {
_subs: Subscriber<Thunk>[] = []
event: TrustedEvent
result = defer<PublishStatusByRelay>()
status: PublishStatusByRelay = {}
details: Record<string, string> = {}
results: PublishResultsByRelay = {}
complete = defer<void>()
controller = new AbortController()
constructor(readonly options: ThunkOptions) {
this.event = prepEvent(options.event)
for (const relay of options.relays) {
this.status[relay] = PublishStatus.Sending
this.results[relay] = {
relay,
status: PublishStatus.Sending,
detail: "sending...",
}
}
this.controller.signal.addEventListener("abort", () => {
for (const relay of options.relays) {
this._setAborted(relay)
this._setAborted({
relay,
status: PublishStatus.Aborted,
detail: "aborted",
})
}
})
}
@@ -82,32 +95,33 @@ export class Thunk {
}
}
_fail(message: string) {
_fail(detail: string) {
for (const relay of this.options.relays) {
this.status[relay] = PublishStatus.Failure
this.details[relay] = message
this.results[relay] = {
relay,
status: PublishStatus.Failure,
detail: detail,
}
}
this._notify()
}
_setPending(relay: string) {
this.options.onPending?.(relay)
this.status[relay] = PublishStatus.Pending
_setPending = (result: PublishResult) => {
this.options.onPending?.(result)
this.results[result.relay] = result
this._notify()
}
_setTimeout(relay: string) {
this.options.onTimeout?.(relay)
this.status[relay] = PublishStatus.Timeout
this.details[relay] = "Publish timed out"
_setTimeout = (result: PublishResult) => {
this.options.onTimeout?.(result)
this.results[result.relay] = result
this._notify()
}
_setAborted(relay: string) {
this.options.onAborted?.(relay)
this.status[relay] = PublishStatus.Aborted
this.details[relay] = "Publish was aborted"
_setAborted = (result: PublishResult) => {
this.options.onAborted?.(result)
this.results[result.relay] = result
this._notify()
}
@@ -159,38 +173,30 @@ export class Thunk {
}
// Send it off
this.result.resolve(
await publish({
...this.options,
event: signedEvent,
onSuccess: (message: string, relay: string) => {
tracker.track(signedEvent.id, relay)
this.options.onSuccess?.(message, relay)
this.status[relay] = PublishStatus.Success
this.details[relay] = message
this._notify()
},
onFailure: (message: string, relay: string) => {
this.options.onFailure?.(message, relay)
this.status[relay] = PublishStatus.Failure
this.details[relay] = message
this._notify()
},
onPending: (relay: string) => {
this._setPending(relay)
},
onTimeout: (relay: string) => {
this._setTimeout(relay)
},
onAborted: (relay: string) => {
this._setAborted(relay)
},
onComplete: () => {
this.options.onComplete?.()
this._subs = []
},
}),
)
await publish({
...this.options,
event: signedEvent,
onSuccess: (result: PublishResult) => {
tracker.track(signedEvent.id, result.relay)
this.options.onSuccess?.(result)
this.results[result.relay] = result
this._notify()
},
onFailure: (result: PublishResult) => {
this.options.onFailure?.(result)
this.results[result.relay] = result
this._notify()
},
onPending: this._setPending,
onTimeout: this._setTimeout,
onAborted: this._setAborted,
onComplete: (result: PublishResult) => {
this.options.onComplete?.(result)
this._subs = []
},
})
this.complete.resolve()
}
subscribe(subscriber: Subscriber<Thunk>) {
@@ -207,8 +213,7 @@ export class Thunk {
export class MergedThunk {
_subs: Subscriber<MergedThunk>[] = []
status: PublishStatusByRelay = {}
details: Record<string, string> = {}
results: PublishResultsByRelay = {}
constructor(readonly thunks: Thunk[]) {
const {Aborted, Failure, Timeout, Pending, Sending, Success} = PublishStatus
@@ -216,16 +221,14 @@ export class MergedThunk {
for (const thunk of thunks) {
thunk.subscribe($thunk => {
this.status = {}
this.details = {}
this.results = {}
for (const relay of relays) {
for (const status of [Aborted, Failure, Timeout, Pending, Sending, Success]) {
const thunk = thunks.find(t => t.status[relay] === status)
const thunk = thunks.find(t => t.results[relay]?.status === status)
if (thunk) {
this.status[relay] = thunk.status[relay]!
this.details[relay] = thunk.details[relay]!
this.results[relay] = thunk.results[relay]!
}
}
}
@@ -271,9 +274,9 @@ export const getThunkUrlsWithStatus = (
) => {
statuses = ensurePlural(statuses)
return Object.entries(thunk.status)
.filter(([_, status]) => statuses.includes(status))
.map(nth(0))
return Object.entries(thunk.results)
.filter(([_, {status}]) => statuses.includes(status))
.map(nth(0)) as string[]
}
export const getCompleteThunkUrls = (thunk: AbstractThunk) =>
@@ -299,9 +302,9 @@ export const thunkIsComplete = (thunk: AbstractThunk) =>
// Thunk errors
export const getThunkError = (thunk: Thunk) => {
for (const [relay, status] of Object.entries(thunk.status)) {
for (const [_, {status, detail}] of Object.entries(thunk.results)) {
if (status === PublishStatus.Failure) {
return thunk.details[relay]
return detail
}
}
packages/content/package.json
+1 -1
View File
@@ -1,6 +1,6 @@
{
"name": "@welshman/content",
"version": "0.5.0",
"version": "0.5.1",
"author": "hodlbod",
"license": "MIT",
"description": "A collection of utilities for parsing nostr note content.",
packages/editor/package.json
+1 -1
View File
@@ -1,6 +1,6 @@
{
"name": "@welshman/editor",
"version": "0.5.0",
"version": "0.5.1",
"author": "hodlbod",
"license": "MIT",
"description": "A batteries-included nostr editor.",
packages/feeds/package.json
+1 -1
View File
@@ -1,6 +1,6 @@
{
"name": "@welshman/feeds",
"version": "0.5.0",
"version": "0.5.1",
"author": "hodlbod",
"license": "MIT",
"description": "Utilities for building dynamic nostr feeds.",
packages/lib/package.json
+1 -1
View File
@@ -1,6 +1,6 @@
{
"name": "@welshman/lib",
"version": "0.5.0",
"version": "0.5.1",
"author": "hodlbod",
"license": "MIT",
"description": "A collection of utilities.",
packages/net/__tests__/publish.test.ts
+26 -6
View File
@@ -1,5 +1,5 @@
import {describe, expect, it, vi, beforeEach, afterEach} from "vitest"
import {publishOne, publish} from "../src/publish"
import {publishOne, publish, PublishStatus} from "../src/publish"
import {MockAdapter} from "../src/adapter"
import {ClientMessageType} from "../src/message"
import {makeEvent} from "@welshman/util"
@@ -40,7 +40,11 @@ describe("publishOne", () => {
await vi.runAllTimers()
expect(successSpy).toHaveBeenCalledWith("hi")
expect(successSpy).toHaveBeenCalledWith({
relay: "1",
detail: "hi",
status: PublishStatus.Success,
})
expect(failureSpy).not.toHaveBeenCalled()
expect(completeSpy).toHaveBeenCalled()
})
@@ -72,7 +76,11 @@ describe("publishOne", () => {
await vi.runAllTimers()
expect(successSpy).not.toHaveBeenCalled()
expect(failureSpy).toHaveBeenCalledWith("hi")
expect(failureSpy).toHaveBeenCalledWith({
relay: "1",
detail: "hi",
status: PublishStatus.Failure,
})
expect(completeSpy).toHaveBeenCalled()
})
@@ -196,9 +204,21 @@ describe("publish", () => {
await vi.runAllTimersAsync()
expect(successSpy).toHaveBeenCalledWith("hi", "1")
expect(failureSpy).toHaveBeenCalledWith("hi", "2")
expect(successSpy).toHaveBeenCalledWith({
relay: "1",
status: PublishStatus.Success,
detail: "hi",
})
expect(failureSpy).toHaveBeenCalledWith({
relay: "2",
status: PublishStatus.Failure,
detail: "hi",
})
expect(completeSpy).toHaveBeenCalledTimes(1)
expect(timeoutSpy).toHaveBeenCalledWith("3")
expect(timeoutSpy).toHaveBeenCalledWith({
relay: "3",
status: PublishStatus.Timeout,
detail: "timed out",
})
})
})
packages/net/package.json
+1 -1
View File
@@ -1,6 +1,6 @@
{
"name": "@welshman/net",
"version": "0.5.0",
"version": "0.5.1",
"author": "hodlbod",
"license": "MIT",
"description": "Utilities for connecting with nostr relays.",
packages/net/src/publish.ts
+66 -66
View File
@@ -1,3 +1,4 @@
import {fromPairs} from "@welshman/lib"
import {SignedEvent} from "@welshman/util"
import {RelayMessage, ClientMessageType, isRelayOk} from "./message.js"
import {AdapterEvent, AdapterContext, getAdapter} from "./adapter.js"
@@ -14,6 +15,7 @@ export enum PublishStatus {
export type PublishResult = {
status: PublishStatus
detail: string
relay: string
}
export type PublishOneOptions = {
@@ -22,26 +24,30 @@ export type PublishOneOptions = {
signal?: AbortSignal
timeout?: number
context?: AdapterContext
onSuccess?: (detail: string) => void
onFailure?: (detail: string) => void
onPending?: () => void
onTimeout?: () => void
onAborted?: () => void
onComplete?: () => void
onSuccess?: (result: PublishResult) => void
onFailure?: (result: PublishResult) => void
onPending?: (result: PublishResult) => void
onTimeout?: (result: PublishResult) => void
onAborted?: (result: PublishResult) => void
onComplete?: (result: PublishResult) => void
}
export const publishOne = (options: PublishOneOptions) =>
new Promise(resolve => {
new Promise<PublishResult>(resolve => {
const adapter = getAdapter(options.relay, options.context)
let status = PublishStatus.Pending
const result = {
relay: options.relay,
status: PublishStatus.Pending,
detail: "",
}
options.onPending?.()
options.onPending?.(result)
const cleanup = () => {
options.onComplete?.()
options.onComplete?.(result)
adapter.cleanup()
resolve(status)
resolve(result)
}
adapter.on(AdapterEvent.Receive, (message: RelayMessage, url: string) => {
@@ -51,11 +57,15 @@ export const publishOne = (options: PublishOneOptions) =>
if (id !== options.event.id) return
if (ok) {
status = PublishStatus.Success
options.onSuccess?.(detail)
result.status = PublishStatus.Success
result.detail = detail
options.onSuccess?.(result)
} else {
status = PublishStatus.Failure
options.onFailure?.(detail)
result.status = PublishStatus.Failure
result.detail = detail
options.onFailure?.(result)
}
cleanup()
@@ -63,18 +73,22 @@ export const publishOne = (options: PublishOneOptions) =>
})
options.signal?.addEventListener("abort", () => {
if (status === PublishStatus.Pending) {
status = PublishStatus.Aborted
options.onAborted?.()
if (result.status === PublishStatus.Pending) {
result.status = PublishStatus.Aborted
result.detail = "aborted"
options.onAborted?.(result)
}
cleanup()
})
setTimeout(() => {
if (status === PublishStatus.Pending) {
status = PublishStatus.Timeout
options.onTimeout?.()
if (result.status === PublishStatus.Pending) {
result.status = PublishStatus.Timeout
result.detail = "timed out"
options.onTimeout?.(result)
}
cleanup()
@@ -83,7 +97,7 @@ export const publishOne = (options: PublishOneOptions) =>
adapter.send([ClientMessageType.Event, options.event])
})
export type PublishStatusByRelay = Record<string, PublishStatus>
export type PublishResultsByRelay = Record<string, PublishResult>
export type PublishOptions = {
event: SignedEvent
@@ -91,17 +105,16 @@ export type PublishOptions = {
signal?: AbortSignal
timeout?: number
context?: AdapterContext
onSuccess?: (detail: string, relay: string) => void
onFailure?: (detail: string, relay: string) => void
onPending?: (relay: string) => void
onTimeout?: (relay: string) => void
onAborted?: (relay: string) => void
onComplete?: () => void
onSuccess?: (result: PublishResult) => void
onFailure?: (result: PublishResult) => void
onPending?: (result: PublishResult) => void
onTimeout?: (result: PublishResult) => void
onAborted?: (result: PublishResult) => void
onComplete?: (result: PublishResult) => void
}
export const publish = async (options: PublishOptions) => {
export const publish = async (options: PublishOptions): Promise<PublishResultsByRelay> => {
const {event, timeout, signal, context} = options
const status: PublishStatusByRelay = {}
const completed = new Set<string>()
const relays = new Set(options.relays)
@@ -109,44 +122,31 @@ export const publish = async (options: PublishOptions) => {
console.warn("Non-unique relays passed to publish")
}
await Promise.all(
options.relays.map(relay =>
publishOne({
event,
relay,
signal,
timeout,
context,
onSuccess: (detail: string) => {
status[relay] = PublishStatus.Success
options.onSuccess?.(detail, relay)
},
onFailure: (detail: string) => {
status[relay] = PublishStatus.Failure
options.onFailure?.(detail, relay)
},
onPending: () => {
status[relay] = PublishStatus.Pending
options.onPending?.(relay)
},
onTimeout: () => {
status[relay] = PublishStatus.Timeout
options.onTimeout?.(relay)
},
onAborted: () => {
status[relay] = PublishStatus.Aborted
options.onAborted?.(relay)
},
onComplete: () => {
completed.add(relay)
return fromPairs(
await Promise.all(
options.relays.map(async relay => {
const result = await publishOne({
event,
relay,
signal,
timeout,
context,
onSuccess: options.onSuccess,
onFailure: options.onFailure,
onPending: options.onPending,
onTimeout: options.onTimeout,
onAborted: options.onAborted,
onComplete: (result: PublishResult) => {
completed.add(relay)
if (completed.size === relays.size) {
options.onComplete?.()
}
},
if (completed.size === relays.size) {
options.onComplete?.(result)
}
},
})
return [relay, result]
}),
),
)
return status
}
packages/relay/package.json
+1 -1
View File
@@ -1,6 +1,6 @@
{
"name": "@welshman/relay",
"version": "0.5.0",
"version": "0.5.1",
"author": "hodlbod",
"license": "MIT",
"description": "An in-memory nostr relay implementation.",
packages/router/package.json
+1 -1
View File
@@ -1,6 +1,6 @@
{
"name": "@welshman/router",
"version": "0.5.0",
"version": "0.5.1",
"author": "hodlbod",
"license": "MIT",
"description": "A collection of utilities for nostr relay selection.",
packages/signer/package.json
+1 -1
View File
@@ -1,6 +1,6 @@
{
"name": "@welshman/signer",
"version": "0.5.0",
"version": "0.5.1",
"author": "hodlbod",
"license": "MIT",
"description": "A nostr signer implemenation supporting several login methods.",
packages/store/package.json
+1 -1
View File
@@ -1,6 +1,6 @@
{
"name": "@welshman/store",
"version": "0.5.0",
"version": "0.5.1",
"author": "hodlbod",
"license": "MIT",
"description": "A collection of utilities based on svelte/store for use with welshman",
packages/util/package.json
+1 -1
View File
@@ -1,6 +1,6 @@
{
"name": "@welshman/util",
"version": "0.5.0",
"version": "0.5.1",
"author": "hodlbod",
"license": "MIT",
"description": "A collection of nostr-related utilities.",
pnpm-lock.yaml
Generated
-17
View File
@@ -20,9 +20,6 @@ importers:
eslint-plugin-prettier:
specifier: ~5.2.5
version: 5.2.6(eslint-config-prettier@10.1.1(eslint@9.23.0))(eslint@9.23.0)(prettier@3.5.3)
fake-indexeddb:
specifier: ^6.0.0
version: 6.0.0
globals:
specifier: ~16.0.0
version: 16.0.0
@@ -95,9 +92,6 @@ importers:
fuse.js:
specifier: ^7.0.0
version: 7.1.0
idb:
specifier: ^8.0.0
version: 8.0.2
svelte:
specifier: ^4.2.18
version: 4.2.19
@@ -1698,10 +1692,6 @@ packages:
resolution: {integrity: sha512-/kP8CAwxzLVEeFrMm4kMmy4CCDlpipyA7MYLVrdJIkV0fYF0UaigQHRsxHiuY/GEea+bh4KSv3TIlgr+2UL6bw==}
engines: {node: '>=12.0.0'}
fake-indexeddb@6.0.0:
resolution: {integrity: sha512-YEboHE5VfopUclOck7LncgIqskAqnv4q0EWbYCaxKKjAvO93c+TJIaBuGy8CBFdbg9nKdpN3AuPRwVBJ4k7NrQ==}
engines: {node: '>=18'}
fast-deep-equal@3.1.3:
resolution: {integrity: sha512-f3qQ9oQy9j2AhBe/H9VC91wLmKBCCU/gDOnKNAYG5hswO7BLKj09Hc5HYNz9cGI++xlpDCIgDaitVs03ATR84Q==}
@@ -1805,9 +1795,6 @@ packages:
engines: {node: '>=18'}
hasBin: true
idb@8.0.2:
resolution: {integrity: sha512-CX70rYhx7GDDQzwwQMDwF6kDRQi5vVs6khHUumDrMecBylKkwvZ8HWvKV08AGb7VbpoGCWUQ4aHzNDgoUiOIUg==}
ignore@5.3.2:
resolution: {integrity: sha512-hsBTNUqQTDwkWtcdYI2i06Y/nUBEsNEDJKjWdigLvegy8kDuJAS8uRlpkkcQpyEXL0Z/pjDy5HBmMjRCJ2gq+g==}
engines: {node: '>= 4'}
@@ -3901,8 +3888,6 @@ snapshots:
expect-type@1.2.1: {}
fake-indexeddb@6.0.0: {}
fast-deep-equal@3.1.3: {}
fast-diff@1.3.0: {}
@@ -4011,8 +3996,6 @@ snapshots:
husky@9.1.7: {}
idb@8.0.2: {}
ignore@5.3.2: {}
import-fresh@3.3.1:
vitest.config.ts
-1
View File
@@ -4,7 +4,6 @@ import {defineConfig} from "vitest/config"
export default defineConfig({
test: {
environment: "happy-dom",
setupFiles: "./vitest.setup.ts",
include: ["packages/**/*.test.ts"],
coverage: {
provider: "v8",