blossom_upload_queue_shim_for_ndk 0.12.0

Offline-first Blossom upload queue for ndk apps. Persists per-upload metadata to a pluggable queue store (sembast included), reads blob bytes from a blossom_cache, and retries until every targeted ser [...]

blossom_upload_queue_shim_for_ndk

Offline-first Blossom upload queue for apps built on the ndk package.

A plain Blossom upload ships a blob to a set of servers once. If every server is unreachable (flaky network, app backgrounded, process killed mid-upload), the blob is gone from the caller's perspective. This package uploads blobs to Blossom servers itself, signing with the accounts NDK holds, and adds:

• Local persistence first. The queue entry is committed to a pluggable queue store and the bytes are pinned in a blossom_cache before any network attempt. upload() returns once persistence is durable; delivery happens in the background and survives restarts.
• 100 % delivery guarantee. An entry is only marked delivered once every targeted server has returned success: true. Partial success keeps the entry pending and retries the missing servers. The one exception is a server refusing the blob for good (too large, type not accepted): it is no longer retried and the entry ends up rejected. See Refusals.
• Monotonic ack history. A server that has acked never un-acks. A delivered entry never silently flips back to pending due to a transient server outage.
• No auto-deletion. Delivered entries stay in the store and can be re-uploaded later, for instance to a freshly discovered server.

Install

dependencies:
  blossom_upload_queue_shim_for_ndk: ^0.12.0
  blossom_cache: ^0.5.0
  ndk: '>=0.10.0-dev.7 <0.12.0'
  idb_shim: ^2.9.2
  sembast: ^3.8.7 # only to use SembastUploadQueueStore

Quick start

import 'dart:typed_data';

import 'package:blossom_cache/blossom_cache.dart';
import 'package:blossom_upload_queue_shim_for_ndk/blossom_upload_queue_shim_for_ndk.dart';
import 'package:blossom_upload_queue_shim_for_ndk/sembast_store.dart';
import 'package:idb_shim/idb_io.dart';
import 'package:ndk/ndk.dart';
import 'package:sembast/sembast_io.dart';

Future<void> main() async {
  final db = await databaseFactoryIo.openDatabase('blossom_uploads.db');
  final store = SembastUploadQueueStore(db);
  final cache = await IdbBlossomCache.open(factory: idbFactorySembastIo);

  final ndk = Ndk(
    NdkConfig(eventVerifier: Bip340EventVerifier(), cache: MemCacheManager()),
  );

  final outbox = OfflineBlossomUpload.withNdk(ndk, cache: cache, store: store);
  outbox.start();

  // Bytes go in the cache first, with a caller-supplied sha256.
  await cache.put(mySha256, myBytes, type: 'image/png');

  // Returns as soon as the queue entry is persisted. Delivery is now the
  // shim's responsibility.
  await outbox.upload(
    sha256: mySha256,
    serverSet: const UrlSet.explicit([
      'https://blossom.primal.net',
      'https://cdn.satellite.earth',
    ]),
  );
}

Queue storage

The shim keeps its queue metadata in an UploadQueueStore. Blob bytes stay in the BlossomCache. Two stores ship with the package:

• SembastUploadQueueStore(db, storeName: 'blob_uploads'), from package:blossom_upload_queue_shim_for_ndk/sembast_store.dart. Open db with sembast_io on native targets, sembast_web in the browser, or sembast_sqflite on top of SQLite. Several queues can share one database under different store names.
• InMemoryUploadQueueStore(), which loses everything on restart. Use it in tests, or as a starting point for your own store.

To keep the queue in your app's own database, implement UploadQueueStore. QueuedBlobUpload.toMap() and QueuedBlobUpload.fromMap() give a JSON-compatible form of each record. The shim relies on three guarantees:

• update() is atomic: no other write to the same key may interleave between the read and the write.
• watch() and watchPending() emit the current state on listen, then after every change.
• name stays stable across restarts. The shim pins pending blobs under a holder derived from it, so renaming a queue orphans those pins, and two queues sharing a cache need distinct names.

Semantics

upload(sha256:, serverSet:, contentType:, pubkey:)

Persists a queue entry for the blob identified by sha256 and schedules an immediate attempt to push it to every server serverSet resolves to. List them with UrlSet.explicit([...]), or describe them: see Server sets. The blob bytes must already live in the BlossomCache passed to the constructor. upload() looks them up via cache.head(sha256) and throws StateError if absent. URLs are normalized (lowercased, trailing / stripped) before storage.

If a record with the same sha256 and pubkey already exists, the server lists are merged. deliveredAt is preserved if every server in the merged list has already acked or refused the blob for good; otherwise the entry is demoted to pending so the missing servers get pushed. A serverSet still to look up always demotes it, until the set resolves.

If no contentType is provided, the shim reads it from the cache descriptor.

pubkey is optional and binds the entry to a nostr account: see Account binding. Leave it out and every attempt is signed by a throwaway key.

enqueue(sha256:, serverSet:, contentType:, pubkey:)

Same as upload(), without the immediate attempt. The entry is due at once, so the next periodic tick, retryNow() or sync() sends it. Without start(), only retryNow() and sync() do: see Sending on demand.

retryNow()

Forces an immediate scan of due entries, bypassing the online check. Use it as an explicit override (e.g. when the user pulls to refresh). Entries waiting out their backoff are left for later; use sync() to attempt them too.

sync({String? pubkey, int concurrency = 4})

Attempts every undelivered entry once, plus those awaiting a reupload(), ignoring both the backoff and the online check. Unlike retryNow(), it completes only when every attempt has finished, and returns a SyncReport:

• delivered: every targeted server has acked.
• failed: attempted, but at least one server still owes an ack. See lastErrors, or resolutionError while a server set is unresolved.
• rejected: settled, but at least one server refused the blob for good. See rejectedServers.
• skipped: not attempted, because the entry's account cannot sign right now.

At most concurrency entries are attempted at once, since each attempt loads its blob into memory. An attempt already in flight is awaited, not duplicated. Pass pubkey to sync one account's entries only.

Sending on demand

An app that only sends when asked, such as a CLI with add and sync commands, never calls start(). Each command is a short-lived process:

// mycli add photo.png
await cache.put(bytes, sha256: sha, type: 'image/png');
await outbox.enqueue(
  sha256: sha,
  serverSet: UrlSet.blossom(pubkey),
  pubkey: pubkey,
);
await outbox.dispose();

// mycli sync
final report = await outbox.sync();
for (final entry in report.failed) {
  print('${entry.sha256}: ${entry.lastErrors}');
}
await outbox.dispose();

With withNdk, an account-bound entry ends up in skipped unless its account is logged in to ndk with a signer when sync() runs.

Server sets

serverSet is a UrlSet from url_set_for_ndk, re-exported by this package. UrlSet.explicit lists servers; the other leaves describe them, to be looked up later:

await outbox.enqueue(
  sha256: sha,
  pubkey: pubkey,
  serverSet: UrlSet.fallback([
    UrlSet.blossom(pubkey), // the account's kind 10063 (BUD-03)
    UrlSet.explicit(['https://blossom.primal.net']),
  ]),
);

Only Blossom leaves are accepted: explicit, blossom, union and fallback. A relay leaf such as UrlSet.dm throws ArgumentError.

A set that needs a lookup is stored as is, so enqueue() works offline even when the server list was never fetched. The worker resolves it before the first upload:

• withNdk reads kind 10063 through NDK, cache first, on the owner's NIP-65 write relays and serverListDiscoveryRelays (default defaultIndexerRelays). With the default constructor, pass a serverListFn. Without one, a set that needs a lookup throws StateError.
• Resolution runs once. The servers it yields are frozen into QueuedBlobUpload.servers; a later change to the list does not move the entry. Use reupload(sha256, server: ...) to add a server afterwards.
• Until then the entry is pending, its blob pinned, and nothing is uploaded. A lookup that is unavailable is retried with backoff, and so is a set that yields no server at all: the account may publish its list later. QueuedBlobUpload.pendingServerSet holds the set and resolutionError says why it is still pending.
• fallback only moves to the next set when a list is known to be missing or empty. An unavailable lookup stops it, so an offline device never uploads to the defaults while the account's own servers might exist.

Connectivity awareness

OfflineBlossomUpload.withNdk() subscribes to ndk.connectivity.relayConnectivityChanges and pauses the periodic retry timer while no public relay is connected. Blossom has no dedicated connectivity stream, so the shim treats "a relay is reachable" as a proxy for "the device has internet." Loopback addresses, RFC1918 IPv4, ULA/link-local IPv6, and mDNS .local names are excluded from the "is online" computation so a local dev relay cannot mask a real outage.

For non-NDK setups, pass any Stream<bool> onlineSignal to the default constructor:

OfflineBlossomUpload(
  cache: cache,
  store: store,
  onlineSignal: yourConnectivityStream, // true while online, false otherwise
);

If you don't pass anything, the shim assumes it is always online and the periodic timer runs unconditionally.

reupload(sha256, {String? server, String? pubkey})

ackedServers and deliveredAt are monotonic. reupload never rewrites the past; it queues a one-shot push via a transient forcedServers override that the next attempt consumes.

• reupload(sha256): schedules an immediate push to every server in the entry's servers list, including those that already acked. Useful when you suspect a server dropped your blob. Acks and deliveredAt are preserved regardless of the new attempt's outcome.
• reupload(sha256, server: s): pushes to that single server. If s is new to the entry, it joins the target list and the entry is demoted to pending until s acks. If s was already there, the historical state is preserved.

Both forms clear the refusals of the servers they push to, so a server that refused the blob is tried again, and the entry is pending until it answers.

pubkey must match the one the entry was queued with.

Account binding

Pass pubkey to upload() to bind an entry to a nostr account:

• Records are keyed by (pubkey, sha256), so the same blob queued by two accounts yields two independent entries. Account-less entries keep the bare sha256 key, so existing databases keep working. get(), watch() and reupload() take an optional pubkey, watchPending() too.
• Every attempt is signed by that account (with withNdk, by its signer in ndk.accounts), whoever is logged in. A retry firing hours later still uploads under the queueing identity.
• If that account can no longer sign (logged out, or read-only), the entry is deferred untouched rather than counted as a failed attempt.
• A signature that comes back under another pubkey fails the attempt: the blob is never sent under the wrong identity.

With the default constructor, pass signEvent to sign as the entry's pubkey, and canSignFor to get the deferral. Without signEvent, upload() and enqueue() with a pubkey throw StateError.

Removing entries

• remove(sha256, pubkey:) deletes that one record, delivered or not. An attempt already in flight is aborted, and the record stays gone.
• clearLocalAccountData(pubkey:) deletes every record bound to that account, delivered or not. Other accounts and account-less records are untouched.
• clearAllLocalData() deletes every record.

Blob bytes stay in the cache either way; only shim-owned pins are released.

Pin ownership

The shim pins the blob in the BlossomCache while an upload is pending and releases the pin once the entry is settled, delivered or rejected. It pins under its own holder, blossom_upload_queue_shim:<name> where <name> is the store's name, suffixed with :<pubkey> for account-bound entries, so each record owns exactly one pin. It never touches pins held by anyone else: pin the blob yourself, before or after upload(), and your pin survives delivery.

A blob whose bytes are deleted from the cache while still pending will have its next attempt fail with lastErrors[server] = 'blob bytes missing from cache'. The entry stays in the store; the network call is not made.

What "success" means

The full target set must ack. Each attempt sends the blob to every target at once, one request per server, so per-server failures do not affect the others. Each answer is saved as soon as it comes back, so a slow server does not hold back the acks of the others. The shim does not mirror (BUD-04) from one server to another: that would introduce server-to-server dependencies that complicate the model.

Refusals

A server answering 413 (too large) or 415 (type not accepted) would give the same answer on every attempt, so the shim stops sending it the blob. The server moves to QueuedBlobUpload.rejectedServers, with the X-Reason it gave if any. Every other failure stays retryable, 401, 402 and 403 included: the account or the server's terms may change.

Once every server has acked or refused, the entry is settled. With at least one refusal its status is rejected rather than delivered: deliveredAt is set, the pin is released, the entry leaves watchPending(), and sync() reports it in rejected. reupload() tries refused servers again.

Blobs of at least checkUploadMinSize bytes (1 MiB by default) are first checked with BUD-06 (HEAD /upload), so a server that would refuse one never receives its bytes. A server may also cut the connection on an oversized upload rather than answer it, and only the check catches that. It is a hint: a server without BUD-06, a check that fails, or one without an answer within checkUploadTimeout (30 s by default, then aborted) lets the upload go ahead. The check carries the same authorization as the upload, so it costs no extra signature.

Requests

The shim talks to Blossom servers itself, over package:http, not through ndk.blossom. Each attempt:

1. Signs one kind 24242 event (BUD-11): t set to upload, x set to the blob's sha256, an expiration 24 hours ahead, and one server tag per target domain, so the token is only valid for this blob on these servers. It travels in Authorization: Nostr <token>, base64url without padding.
2. For a blob of at least checkUploadMinSize bytes, sends HEAD /upload to every target with X-SHA-256, X-Content-Length and X-Content-Type (BUD-06).
3. Sends PUT /upload to the targets that did not refuse, with X-SHA-256 and Content-Type (BUD-02). Any 2xx is an ack.

An account-bound entry is signed by its account, an account-less one by a key generated for that attempt. With a remote signer, that is one prompt per attempt, however many servers it reaches.

Neither the signature nor the upload has a timeout. A remote signer may wait for its user to approve, and a slow link, such as Tor, may take hours to send a large blob: any deadline would assume a minimum speed nobody knows. An upload ends when the network fails it, or when remove(), a clear* method or dispose() aborts it. dispose() does not wait for a pending signature either. An attempt it aborts does not count: the answers it got are kept, and the entry stays due for the next launch.

Pass an httpClient to either constructor to choose the transport, for instance one with a proxy or custom certificates. It stays yours to close. A client the shim opened itself is closed by dispose().

What the shim does NOT do

• It never hashes. The caller provides the sha256. The cache stores by that key. The shim queues by that key.
• It never uses BUD-05 /media. The server would re-encode the blob and change its sha256, breaking content-addressing.
• It never deletes records on its own. Even after full delivery, the entry stays in the store until you call remove() or one of the clear* methods.
• It does not give up on other failures. Only a refusal for good stops it for a server: see Refusals. Anything else, such as a server that is down or asks for payment, is retried forever with exponential backoff. Inspect QueuedBlobUpload.lastErrors and call remove() if needed.

Tuning

OfflineBlossomUpload.withNdk(
  ndk,
  cache: cache,
  store: SembastUploadQueueStore(db, storeName: 'blob_uploads'),
  tickInterval: const Duration(seconds: 30),       // periodic retry scan
  initialBackoff: const Duration(seconds: 5),      // backoff floor
  maxBackoff: const Duration(minutes: 30),         // backoff ceiling
  checkUploadMinSize: 1024 * 1024,                 // BUD-06 check before sending blobs this big
  checkUploadTimeout: const Duration(seconds: 30), // abort a BUD-06 check after this
);

Testing your integration

OfflineBlossomUpload is fully unit-testable without NDK, a network or a database. Pass a MockClient from package:http/testing.dart to the default constructor:

final outbox = OfflineBlossomUpload(
  httpClient: MockClient((request) async => http.Response('{}', 201)),
  signEvent: (event) async => event.copyWith(sig: 'test'),
  cache: await IdbBlossomCache.open(factory: newIdbFactoryMemory()),
  store: InMemoryUploadQueueStore(),
);

The package's own test suite uses this approach; see test/offline_blossom_upload_test.dart. test/upload_queue_store_test.dart checks both stores against the contract above, and is a good template to test your own.

License

MIT