Sterister
d47774ef1c
Two follow-ups to the V4.8.2 duplicate-fan-out fixes Prism filed. 1. `Inbox.acceptBridgeFrame(blob)` + shared 4096-entry msgId LRU. The relay durably stores blobs and pushes them to every active delivery channel; without a cross-channel ack the bridge frame ran first and the next inbox-poll re-dispatched the same blob ~30 s later, tripping on consumed prekeys. Bridge consumers now plumb pushed frames through `acceptBridgeFrame`, which shares the dedup gate + ack path with `pollOnce`. Whichever channel delivers first wins; the other acks-and-skips. Inbox records the msgId before the ack so a parallel poll can't observe an in-flight ack window. 2. `Shade.aliasSession(oldLabel, newLabel)`. First-contact forces the receiver to label the new session by the relay's sender fingerprint hint (`fp:<senderfp>`); the post-decrypt plaintext typically announces the peer's real address. Aliasing moves session, trusted identity, peer-verification, and identity- version under the canonical label. Holds the per-peer mutex on both labels (lexicographic order) so concurrent crypto ops can't observe a half-moved state. Refuses to overwrite an existing session at the new label. Wire change: `IncomingMessage.expiresAt?` now surfaces the relay's expiry so receivers can pass bridge frames straight to `acceptBridgeFrame` without inventing a TTL. Tests cover bridge-then-poll, poll-then-bridge, aliasSession happy path, refuse-to-overwrite, and same-label no-op. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
1730 lines
67 KiB
TypeScript
1730 lines
67 KiB
TypeScript
import type { ShadeEnvelope, StorageProvider } from '@shade/core';
|
||
import {
|
||
ShadeSessionManager,
|
||
ShadeEventEmitter,
|
||
NoSessionError,
|
||
} from '@shade/core';
|
||
import {
|
||
FingerprintGateRegistry,
|
||
type FingerprintGateHandler,
|
||
} from './gates.js';
|
||
import {
|
||
SubtleCryptoProvider,
|
||
MemoryStorage,
|
||
createWorkerCryptoProvider,
|
||
type WorkerCryptoProvider,
|
||
createEncryptStream,
|
||
createDecryptStream,
|
||
type CreateEncryptStreamOptions,
|
||
type CreateDecryptStreamOptions,
|
||
} from '@shade/crypto-web';
|
||
import { encodeEnvelope, decodeEnvelope, inspectEnvelopeType } from '@shade/proto';
|
||
import { ShadeFetchTransport, type KTVerifierOptions } from '@shade/transport';
|
||
import { LightWitness } from '@shade/key-transparency';
|
||
import type { SignedTreeHead, STHWire } from '@shade/key-transparency';
|
||
import {
|
||
TransferEngine,
|
||
ShadeTransferHttpTransport,
|
||
MultiTransportFallback,
|
||
type ITransferTransport,
|
||
type IncomingTransfer,
|
||
type TransferHandle,
|
||
type TransferOptions,
|
||
type TransferSummary,
|
||
type OutboundQueue as OutboundQueueLike,
|
||
type QueuedEventInput,
|
||
} from '@shade/transfer';
|
||
import type { Hono } from 'hono';
|
||
import { BackgroundTasks } from './background.js';
|
||
import {
|
||
exportBackup,
|
||
applyBackupPayload,
|
||
decryptBackup,
|
||
backupToString,
|
||
backupFromString,
|
||
} from './backup.js';
|
||
import { computeFingerprint, deserializeIdentityKeyPair } from '@shade/core';
|
||
import {
|
||
acceptBroadcastEnvelope,
|
||
createBroadcastChannelImpl,
|
||
getBroadcastChannelImpl,
|
||
listBroadcastChannelsImpl,
|
||
maybeHandleControlPlaintext,
|
||
type BroadcastChannel,
|
||
type BroadcastChannelSummary,
|
||
type BroadcastSdkHooks,
|
||
type MessageMeta,
|
||
} from './broadcast.js';
|
||
import type { ResolvedConfig, StorageSpec } from './config.js';
|
||
import {
|
||
ShadeControlChannel,
|
||
ShadeTransferAuthenticator,
|
||
type ControlEnvelopeTransport,
|
||
} from './streams-bridge.js';
|
||
import { createFilesNamespace, type FilesNamespace } from '@shade/files';
|
||
import type { ObservabilityHook } from '@shade/observability';
|
||
import {
|
||
isAllowedThumbnailMime,
|
||
sha256Once,
|
||
THUMBNAIL_MAX_BYTES,
|
||
type StreamFileMetadata,
|
||
type ThumbnailMime,
|
||
} from '@shade/streams';
|
||
import {
|
||
generateThumbnail as generateThumbnailFromBlob,
|
||
type ThumbnailGenerationOptions,
|
||
} from './thumbnail.js';
|
||
|
||
/**
|
||
* V3.9 — extended upload options. The base `TransferOptions` is forwarded
|
||
* verbatim to `@shade/transfer`; the extra fields control the thumbnail
|
||
* companion-stream and never leak past the SDK boundary.
|
||
*/
|
||
/**
|
||
* V3.11 — opt-in WebRTC P2P transport. Pass to `shade.configureWebRTC()`
|
||
* before the engine is built. The shape mirrors `WebRtcConnectionManager`'s
|
||
* options without forcing the SDK to import `@shade/transport-webrtc` at
|
||
* the value layer (the import happens lazily inside `engine()`).
|
||
*/
|
||
export interface ShadeWebRtcConfig {
|
||
/**
|
||
* WebRTC adapter. Use `nativeRtcFactory()` from `@shade/transport-webrtc`
|
||
* in browsers / Deno / Cloudflare Workers; supply your own
|
||
* `IRtcFactory` for Node-class environments (`node-datachannel`, `wrtc`).
|
||
*/
|
||
factory: import('@shade/transport-webrtc').IRtcFactory;
|
||
iceServers?: import('@shade/transport-webrtc').ShadeRtcConfig['iceServers'];
|
||
iceTransportPolicy?: import('@shade/transport-webrtc').ShadeRtcConfig['iceTransportPolicy'];
|
||
bundlePolicy?: import('@shade/transport-webrtc').ShadeRtcConfig['bundlePolicy'];
|
||
/** Default 30s. */
|
||
connectTimeoutMs?: number;
|
||
/** Default 30s. */
|
||
requestTimeoutMs?: number;
|
||
/** Default 4 MiB. */
|
||
backpressureThresholdBytes?: number;
|
||
}
|
||
|
||
/** Live WebRTC runtime returned by `shade.getWebRtcRuntime()`. */
|
||
export interface ShadeWebRtcRuntime {
|
||
signaling: import('@shade/transport-webrtc').WebRtcSignalingChannel;
|
||
manager: import('@shade/transport-webrtc').WebRtcConnectionManager;
|
||
transport: import('@shade/transport-webrtc').WebRtcTransferTransport;
|
||
fallback: MultiTransportFallback;
|
||
/** Internal — wires `engine` into the receiver hooks once it's built. */
|
||
attachEngine(engine: TransferEngine): void;
|
||
/** Tear down the manager, the signaling channel, and any open peer connections. */
|
||
destroy(): void;
|
||
}
|
||
|
||
export interface ShadeUploadOptions extends TransferOptions {
|
||
/**
|
||
* Pre-generated thumbnail bytes + MIME. Use this on server runtimes
|
||
* where in-process image processing is already part of your pipeline,
|
||
* or when you have a richer thumbnail than the auto-generator would
|
||
* produce (e.g. center-cropped, branded watermark, etc.).
|
||
*/
|
||
thumbnail?: { bytes: Uint8Array; mime: ThumbnailMime };
|
||
/**
|
||
* Browser auto-generation. Set to `true` for defaults, or pass a
|
||
* config object. Returns `null` (silently skips the thumbnail) on
|
||
* runtimes lacking `OffscreenCanvas` + `createImageBitmap`.
|
||
*/
|
||
generateThumbnail?: boolean | ThumbnailGenerationOptions;
|
||
}
|
||
|
||
/**
|
||
* The high-level Shade API.
|
||
*
|
||
* Wraps crypto, storage, session management, transport, and optional
|
||
* observer into a single object. Provides magic auto-establish + auto-
|
||
* publish + auto-replenish behavior.
|
||
*/
|
||
export class Shade {
|
||
private readonly crypto = new SubtleCryptoProvider();
|
||
private readonly events = new ShadeEventEmitter();
|
||
private storage!: StorageProvider;
|
||
private manager!: ShadeSessionManager;
|
||
private transport!: ShadeFetchTransport;
|
||
private _background: BackgroundTasks | null = null;
|
||
private address!: string;
|
||
private initialized = false;
|
||
|
||
// Per-address mutex to serialize session establishment under concurrent sends
|
||
private establishing = new Map<string, Promise<void>>();
|
||
// Per-address encrypt queue to serialize ratchet mutations
|
||
private encryptChains = new Map<string, Promise<unknown>>();
|
||
// Per-`from` decrypt queue: serializes incoming receives so two concurrent
|
||
// shade.receive(from, env) calls can't race the ratchet/storage. Without
|
||
// this, parallel deliveries (relay duplicate fan-out, fast pipelined
|
||
// sends) hit `database is locked` (sqlite) or transaction conflicts (IDB)
|
||
// because the underlying StorageProvider isn't required to be a
|
||
// concurrent-safe writer. See V4.8.2 changelog.
|
||
private decryptChains = new Map<string, Promise<unknown>>();
|
||
|
||
// Message handlers — may be sync or async; receive() awaits each. The
|
||
// optional third arg distinguishes direct vs broadcast plaintexts;
|
||
// handlers registered without it work unchanged (V4.6 back-compat).
|
||
private messageHandlers: Array<
|
||
(from: string, plaintext: string, meta?: MessageMeta) => void | Promise<void>
|
||
> = [];
|
||
|
||
// Stream-transfer engine, lazily constructed on first use.
|
||
private transferEngine: TransferEngine | null = null;
|
||
private controlChannel: ShadeControlChannel | null = null;
|
||
private peerBaseUrlResolver: ((peerAddress: string) => Promise<string>) | null = null;
|
||
private envelopeOutboxes: ControlEnvelopeTransport | null = null;
|
||
private transferTransportOverride: ITransferTransport | null = null;
|
||
private transferQueue: OutboundQueueLike | null = null;
|
||
|
||
// `@shade/files` namespace, lazy + memoized.
|
||
private filesNamespace: FilesNamespace | null = null;
|
||
|
||
// V3.12 — light-witness for split-view detection.
|
||
private ktWitness: LightWitness | null = null;
|
||
|
||
// V3.11 WebRTC P2P transport. Lazy-built on first engine() if configured.
|
||
private webrtcConfig: ShadeWebRtcConfig | null = null;
|
||
private webrtcRuntime: ShadeWebRtcRuntime | null = null;
|
||
|
||
// V3.8 Worker-Crypto. Lazy: configured via `configureWorkerCrypto()`,
|
||
// spawned the first time `encryptStream`/`decryptStream` is used.
|
||
private workerCryptoConfig: {
|
||
workerUrl: URL | string;
|
||
idleTimeoutMs?: number;
|
||
} | null = null;
|
||
private workerCrypto: WorkerCryptoProvider | null = null;
|
||
private workerCryptoBoot: Promise<WorkerCryptoProvider> | null = null;
|
||
|
||
// V3.3 fingerprint gates. Created in `initialize()` once storage is up.
|
||
private gates!: FingerprintGateRegistry;
|
||
|
||
constructor(private readonly config: ResolvedConfig) {}
|
||
|
||
/**
|
||
* Initialize the SDK:
|
||
* 1. Resolve storage backend
|
||
* 2. Create session manager + generate identity if needed
|
||
* 3. Create transport
|
||
* 4. Generate initial one-time prekeys
|
||
* 5. Register with prekey server
|
||
* 6. Start background tasks
|
||
*/
|
||
async initialize(): Promise<void> {
|
||
if (this.initialized) return;
|
||
|
||
// Step 1: Storage
|
||
this.storage = await resolveStorage(this.config.storage);
|
||
|
||
// Step 2: Session manager with event bus attached
|
||
this.manager = new ShadeSessionManager(this.crypto, this.storage, {
|
||
events: this.events,
|
||
...(this.config.observability !== undefined
|
||
? { observability: this.config.observability }
|
||
: {}),
|
||
});
|
||
await this.manager.initialize();
|
||
|
||
// Step 3: Address (user-provided or persisted UUID)
|
||
this.address = this.config.address ?? (await resolveAddress(this.storage));
|
||
|
||
// Step 4: Transport with our signing key
|
||
const identity = await this.storage.getIdentityKeyPair();
|
||
if (!identity) throw new Error('Identity not available after initialize');
|
||
|
||
// V3.12 — wire up KT verifier + light-witness if configured.
|
||
let ktOpts: KTVerifierOptions | undefined;
|
||
if (this.config.keyTransparency) {
|
||
const baseUrl = this.config.prekeyServer;
|
||
this.ktWitness = new LightWitness({
|
||
crypto: this.crypto,
|
||
logPublicKey: this.config.keyTransparency.logPublicKey,
|
||
maxStaleMs: this.config.keyTransparency.maxStaleMs,
|
||
maxStored: this.config.keyTransparency.witnessMaxStored,
|
||
fetcher: {
|
||
async fetchLatestSTH(): Promise<STHWire> {
|
||
const res = await fetch(`${baseUrl}/v1/kt/sth`);
|
||
if (!res.ok) throw new Error(`KT /sth: ${res.status}`);
|
||
return (await res.json()) as STHWire;
|
||
},
|
||
async fetchConsistencyProof(from, to): Promise<{ proof: string[] }> {
|
||
const res = await fetch(`${baseUrl}/v1/kt/consistency?from=${from}&to=${to}`);
|
||
if (!res.ok) throw new Error(`KT /consistency: ${res.status}`);
|
||
return (await res.json()) as { proof: string[] };
|
||
},
|
||
},
|
||
});
|
||
ktOpts = {
|
||
mode: this.config.keyTransparency.mode,
|
||
logPublicKey: this.config.keyTransparency.logPublicKey,
|
||
maxStaleMs: this.config.keyTransparency.maxStaleMs,
|
||
onObserveSth: async (sth: SignedTreeHead) => {
|
||
if (this.ktWitness) {
|
||
// The fetched STH was already verified by the transport; feed
|
||
// it to the witness for split-view tracking. `observe` may also
|
||
// throw on split-view — we let it propagate to the caller.
|
||
await this.ktWitness.observe(sth);
|
||
}
|
||
},
|
||
};
|
||
}
|
||
|
||
this.transport = new ShadeFetchTransport({
|
||
baseUrl: this.config.prekeyServer,
|
||
crypto: this.crypto,
|
||
signingPrivateKey: identity.signingPrivateKey,
|
||
...(ktOpts ? { keyTransparency: ktOpts } : {}),
|
||
});
|
||
|
||
// Step 5: Initial prekeys + register
|
||
const otpks = await this.manager.generateOneTimePreKeys(20);
|
||
const bundle = await this.manager.createPreKeyBundle();
|
||
|
||
try {
|
||
await this.transport.register(
|
||
this.address,
|
||
this.manager.getPublicIdentity(),
|
||
bundle.signedPreKey,
|
||
otpks,
|
||
);
|
||
} catch (err) {
|
||
console.warn(
|
||
`[Shade] Failed to register with prekey server at ${this.config.prekeyServer}: ${(err as Error).message}. Will retry on next replenish.`,
|
||
);
|
||
}
|
||
|
||
// Step 6: Background tasks
|
||
this._background = new BackgroundTasks(
|
||
this.manager,
|
||
this.transport,
|
||
this.address,
|
||
this.config,
|
||
);
|
||
this._background.start();
|
||
|
||
// Step 7: V3.3 fingerprint gates
|
||
this.gates = new FingerprintGateRegistry(this.storage);
|
||
|
||
this.initialized = true;
|
||
}
|
||
|
||
/** Your identity's safety number (12 groups × 5 digits) */
|
||
get fingerprint(): Promise<string> {
|
||
if (!this.initialized) throw new Error('Not initialized');
|
||
return this.manager.getIdentityFingerprint();
|
||
}
|
||
|
||
/** Your address on the prekey server */
|
||
get myAddress(): string {
|
||
if (!this.initialized) throw new Error('Not initialized');
|
||
return this.address;
|
||
}
|
||
|
||
/**
|
||
* The local device's Ed25519 identity public key (32 bytes).
|
||
*
|
||
* Stable for the lifetime of the identity. After {@link rotate} this
|
||
* reflects the new key; the previous key is preserved in retired-
|
||
* identities storage for the configured grace period.
|
||
*
|
||
* Hand this to your application's backend at enrollment time so it
|
||
* can verify signatures from this device, compute its own safety-
|
||
* number representation, or pin the key for later attestation. Use
|
||
* {@link fingerprint} instead for human side-channel comparison.
|
||
*/
|
||
get identityPublicKey(): Promise<Uint8Array> {
|
||
if (!this.initialized) throw new Error('Not initialized');
|
||
return this.storage.getIdentityKeyPair().then((kp) => {
|
||
if (!kp) throw new Error('Identity not yet generated');
|
||
return kp.signingPublicKey;
|
||
});
|
||
}
|
||
|
||
/**
|
||
* `@shade/files` namespace — high-level entry point for E2EE filesystem
|
||
* RPC. Lazily creates the underlying channel + streams bridges on first
|
||
* access; subsequent accesses return the same instance.
|
||
*
|
||
* ```ts
|
||
* const files = shade.files;
|
||
* const stop = await files.serve({ list: ..., write: ..., ... });
|
||
* const fs = await files.client('bob');
|
||
* await fs.list('/');
|
||
* ```
|
||
*
|
||
* Requires `configureTransfers({ resolveBaseUrl })` to be called first
|
||
* (same as `upload`/`onIncomingTransfer`).
|
||
*/
|
||
get files(): FilesNamespace {
|
||
if (!this.initialized) throw new Error('Not initialized');
|
||
if (this.filesNamespace !== null) return this.filesNamespace;
|
||
// `@shade/files` only imports `Shade` as a type, so the cyclic ESM
|
||
// import is type-only at the value layer — safe to bind synchronously.
|
||
this.filesNamespace = createFilesNamespace(this);
|
||
return this.filesNamespace;
|
||
}
|
||
|
||
/** Internal — exposes the BackgroundTasks for `@shade/files` to wire prune. */
|
||
get background(): BackgroundTasks | null {
|
||
return this._background;
|
||
}
|
||
|
||
/** Access the underlying event emitter (for observer integration) */
|
||
getEvents(): ShadeEventEmitter {
|
||
return this.events;
|
||
}
|
||
|
||
/** Access the underlying session manager (for advanced usage) */
|
||
getManager(): ShadeSessionManager {
|
||
return this.manager;
|
||
}
|
||
|
||
/** Access the underlying transport (for advanced usage) */
|
||
getTransport(): ShadeFetchTransport {
|
||
return this.transport;
|
||
}
|
||
|
||
/**
|
||
* V3.12 — access the configured Key-Transparency light-witness, or
|
||
* `null` when KT was not configured. Useful for surfacing observed
|
||
* STHs to UI / dashboards, or for manual gossip checks against
|
||
* trusted peers.
|
||
*/
|
||
getKTWitness(): LightWitness | null {
|
||
return this.ktWitness;
|
||
}
|
||
|
||
/**
|
||
* Returns the OTel observability hook the SDK was configured with, or
|
||
* `undefined` if observability is off. Used by `@shade/files` and other
|
||
* sub-modules to inherit the same tracer the rest of the SDK uses.
|
||
*/
|
||
getObservability(): ObservabilityHook | undefined {
|
||
return this.config.observability;
|
||
}
|
||
|
||
/**
|
||
* Encrypt a message to a peer. Auto-establishes a session if none exists.
|
||
* Returns the ShadeEnvelope ready to send over any transport.
|
||
*/
|
||
async send(address: string, plaintext: string): Promise<ShadeEnvelope> {
|
||
if (!this.initialized) throw new Error('Not initialized');
|
||
|
||
// Serialize all sends to the same peer: the SessionManager mutates
|
||
// ratchet state in place, and interleaved mutations corrupt it.
|
||
const previous = this.encryptChains.get(address) ?? Promise.resolve();
|
||
const next = previous
|
||
.catch(() => {}) // don't propagate upstream failures to later sends
|
||
.then(async () => {
|
||
try {
|
||
return await this.manager.encrypt(address, plaintext);
|
||
} catch (err) {
|
||
if (!(err instanceof NoSessionError)) throw err;
|
||
await this.ensureSession(address);
|
||
return this.manager.encrypt(address, plaintext);
|
||
}
|
||
});
|
||
this.encryptChains.set(address, next);
|
||
return next as Promise<ShadeEnvelope>;
|
||
}
|
||
|
||
/**
|
||
* Decrypt an incoming envelope and notify registered message handlers.
|
||
* Returns the plaintext.
|
||
*
|
||
* The caller provides the `from` address because the envelope itself
|
||
* doesn't authenticate the sender — that's determined by your transport
|
||
* layer (auth header, WebSocket peer, push notification metadata, etc.).
|
||
*
|
||
* V4.6: when the decrypted plaintext is a broadcast control message
|
||
* (sender-key distribution / revocation), the SDK consumes it
|
||
* internally and returns an empty string; user handlers do NOT fire.
|
||
* Apps therefore see only direct plaintexts here. Broadcast payloads
|
||
* arrive via {@link Shade.acceptBroadcast}.
|
||
*/
|
||
async receive(from: string, envelope: ShadeEnvelope): Promise<string> {
|
||
if (!this.initialized) throw new Error('Not initialized');
|
||
|
||
// Serialize ONLY the ratchet/storage write portion of receive (the
|
||
// call into `manager.decrypt`). Concurrent decrypts race the
|
||
// SessionManager ratchet (mutated in place) and the StorageProvider
|
||
// (not required to be a concurrent-safe writer — `bun:sqlite`
|
||
// throws `database is locked`, IDB throws transaction conflicts).
|
||
// The Prism FR called this out: a relay-duplicated WS fan-out
|
||
// dispatched 8 parallel `shade.receive(from, env)` calls, one won
|
||
// the X3DH prekey race and the other 7 failed with
|
||
// `database is locked` / `one-time prekey not found`. The fix is
|
||
// to queue per-`from` decrypts so the ratchet step is sequential.
|
||
//
|
||
// Crucially the user-visible MESSAGE HANDLERS run *outside* the
|
||
// queue. Streams + file-RPC issue nested `shade.receive` calls for
|
||
// the same peer from inside their handlers (e.g. `stream-end`
|
||
// arrives while a write-RPC is still waiting on chunks); holding
|
||
// the queue across the handler would self-deadlock. The atomic
|
||
// unit we have to protect is just the ratchet+storage step, not
|
||
// the consumer's reaction to it.
|
||
const previous = this.decryptChains.get(from) ?? Promise.resolve();
|
||
const decryptPromise = previous
|
||
.catch(() => undefined) // don't propagate upstream failures
|
||
.then(() => this.manager.decrypt(from, envelope));
|
||
// Store a never-rejecting copy so the next chained receive doesn't
|
||
// see a rejection from this one (we still surface our own rejection
|
||
// to *this* caller via the original `decryptPromise`).
|
||
this.decryptChains.set(from, decryptPromise.catch(() => undefined));
|
||
const plaintext = await decryptPromise;
|
||
|
||
const consumed = await maybeHandleControlPlaintext(
|
||
this.broadcastHooks(),
|
||
from,
|
||
plaintext,
|
||
);
|
||
if (consumed) return '';
|
||
const meta: MessageMeta = { kind: 'direct' };
|
||
for (const handler of this.messageHandlers) {
|
||
try {
|
||
await handler(from, plaintext, meta);
|
||
} catch (err) {
|
||
console.error('[Shade] Message handler threw:', err);
|
||
}
|
||
}
|
||
return plaintext;
|
||
}
|
||
|
||
/**
|
||
* Register a handler for incoming messages. Async handlers are awaited.
|
||
*
|
||
* V4.6: handlers may declare an optional `meta` parameter to discriminate
|
||
* direct (`meta.kind === 'direct'`) from broadcast (`meta.kind === 'broadcast'`)
|
||
* deliveries. Handlers that ignore the third arg keep working unchanged
|
||
* for direct messages.
|
||
*/
|
||
onMessage(
|
||
handler: (from: string, plaintext: string, meta?: MessageMeta) => void | Promise<void>,
|
||
): () => void {
|
||
this.messageHandlers.push(handler);
|
||
return () => {
|
||
this.messageHandlers = this.messageHandlers.filter((h) => h !== handler);
|
||
};
|
||
}
|
||
|
||
// ─── V4.6 Broadcast channels ───────────────────────────────
|
||
|
||
/**
|
||
* Create a new broadcast channel owned by this device. Returns a
|
||
* handle for adding/removing members, encrypting a single payload,
|
||
* and rotating on revocation. The channel id is opaque, stable
|
||
* across `shutdown()` / re-open, and persisted via the configured
|
||
* `StorageProvider`.
|
||
*/
|
||
async createBroadcastChannel(opts: { label?: string } = {}): Promise<BroadcastChannel> {
|
||
if (!this.initialized) throw new Error('Not initialized');
|
||
return createBroadcastChannelImpl(this.broadcastHooks(), opts);
|
||
}
|
||
|
||
/**
|
||
* Look up an existing sender-side broadcast channel by id. Returns
|
||
* `null` when the id is unknown OR when this device only holds a
|
||
* receiver-side copy (the receiver path uses `onMessage` for delivery
|
||
* — there is no app-facing handle on the receive side).
|
||
*/
|
||
async getBroadcastChannel(channelId: string): Promise<BroadcastChannel | null> {
|
||
if (!this.initialized) throw new Error('Not initialized');
|
||
return getBroadcastChannelImpl(this.broadcastHooks(), channelId);
|
||
}
|
||
|
||
/**
|
||
* Snapshot of every broadcast channel persisted on this device,
|
||
* including receiver-side channels that we joined. Useful for
|
||
* rebuilding UI state on startup.
|
||
*/
|
||
async listBroadcastChannels(): Promise<readonly BroadcastChannelSummary[]> {
|
||
if (!this.initialized) throw new Error('Not initialized');
|
||
return listBroadcastChannelsImpl(this.broadcastHooks());
|
||
}
|
||
|
||
/**
|
||
* Hand a wire-encoded broadcast envelope (type 0x21) to the SDK.
|
||
* Decrypts via the matching channel, advances the chain, and dispatches
|
||
* the plaintext to `onMessage` handlers with `meta.kind === 'broadcast'`.
|
||
*
|
||
* Stale generations (sender's old chain after a rotation we already
|
||
* received) are silently dropped. Future generations (we haven't seen
|
||
* the rotation distribution yet) throw — the app should ensure the
|
||
* distribution envelope is delivered before the broadcast.
|
||
*/
|
||
async acceptBroadcast(envelope: Uint8Array): Promise<void> {
|
||
if (!this.initialized) throw new Error('Not initialized');
|
||
const result = await acceptBroadcastEnvelope(this.broadcastHooks(), envelope);
|
||
if (result === null) return;
|
||
for (const handler of this.messageHandlers) {
|
||
try {
|
||
await handler(result.meta.kind === 'broadcast' ? result.meta.sender : '', result.plaintext, result.meta);
|
||
} catch (err) {
|
||
console.error('[Shade] Broadcast handler threw:', err);
|
||
}
|
||
}
|
||
}
|
||
|
||
private broadcastHooks(): BroadcastSdkHooks {
|
||
return {
|
||
bilateralSend: (peer, pt) => this.send(peer, pt),
|
||
myAddress: () => this.address,
|
||
crypto: this.crypto,
|
||
storage: this.storage,
|
||
};
|
||
}
|
||
|
||
/** Get a peer's fingerprint (requires an existing session) */
|
||
async getFingerprintFor(address: string): Promise<string> {
|
||
if (!this.initialized) throw new Error('Not initialized');
|
||
return this.manager.getRemoteFingerprint(address);
|
||
}
|
||
|
||
/** Verify a fingerprint matches the pinned identity for an address */
|
||
async verify(address: string, fingerprint: string): Promise<boolean> {
|
||
const remote = await this.getFingerprintFor(address);
|
||
return normalize(remote) === normalize(fingerprint);
|
||
}
|
||
|
||
// ─── V3.3 fingerprint gates ───────────────────────────────
|
||
|
||
/**
|
||
* Register a handler that runs before `upload()` proceeds when the file
|
||
* is at or above `threshold` bytes and the peer is not yet verified.
|
||
* Return `true` to allow + persist the verification, `false` to abort.
|
||
*
|
||
* Default threshold (when this method is never called): 10 MiB.
|
||
*/
|
||
beforeFirstLargeFile(threshold: number, handler: FingerprintGateHandler): void {
|
||
if (!this.initialized) throw new Error('Not initialized');
|
||
this.gates.registerFirstLargeFile(threshold, handler);
|
||
}
|
||
|
||
/**
|
||
* Register a handler that runs before `importBackup()` writes to storage.
|
||
* The handler receives the fingerprint of the identity *embedded in the
|
||
* backup blob*, so the user can OOB-confirm the backup is theirs.
|
||
*/
|
||
beforeBackupImport(handler: FingerprintGateHandler): void {
|
||
if (!this.initialized) throw new Error('Not initialized');
|
||
this.gates.registerBackupImport(handler);
|
||
}
|
||
|
||
/**
|
||
* Register a handler that runs the first time a peer's rotated identity
|
||
* is observed (via `acceptIdentityChange` or X3DH against a new bundle).
|
||
*/
|
||
beforeNewDeviceTrust(handler: FingerprintGateHandler): void {
|
||
if (!this.initialized) throw new Error('Not initialized');
|
||
this.gates.registerNewDeviceTrust(handler);
|
||
}
|
||
|
||
/**
|
||
* Register a handler that runs per-recipient before an inbox fan-out
|
||
* delivery (V3.6). Reserved hook — wired here so apps can register it
|
||
* today and have it active automatically when V3.6 ships.
|
||
*/
|
||
beforeInboxFanout(handler: FingerprintGateHandler): void {
|
||
if (!this.initialized) throw new Error('Not initialized');
|
||
this.gates.registerInboxFanout(handler);
|
||
}
|
||
|
||
/**
|
||
* Mark a peer as verified at their current fingerprint. Call this from
|
||
* your own UI (e.g. after the user scans a QR code or reads the safety
|
||
* number aloud) to satisfy any gate without going through the handler.
|
||
*/
|
||
async markPeerVerified(address: string): Promise<void> {
|
||
if (!this.initialized) throw new Error('Not initialized');
|
||
const fingerprint = await this.manager.getRemoteFingerprint(address);
|
||
await this.gates.markVerified(address, fingerprint, 'user');
|
||
}
|
||
|
||
/**
|
||
* Returns whether `address` has a current verification (fingerprint and
|
||
* identity-version both still match).
|
||
*/
|
||
async isPeerVerified(address: string): Promise<boolean> {
|
||
if (!this.initialized) throw new Error('Not initialized');
|
||
const fingerprint = await this.manager.getRemoteFingerprint(address);
|
||
return this.gates.isVerified(address, fingerprint);
|
||
}
|
||
|
||
/** Drop any persisted verification for `address`. */
|
||
async unmarkPeerVerified(address: string): Promise<void> {
|
||
if (!this.initialized) throw new Error('Not initialized');
|
||
await this.gates.revoke(address);
|
||
}
|
||
|
||
/**
|
||
* Move every per-peer storage row for `oldLabel` (session, trusted
|
||
* identity, peer-verification, identity-version counter) to
|
||
* `newLabel`. Use this when first-contact forced you to label a
|
||
* session by the relay's sender-fingerprint hint
|
||
* (`fp:<hex>` — see `IncomingMessage.from` / `FetchedBlob.from`) and
|
||
* the just-decrypted plaintext announces the peer's canonical
|
||
* address: alias once and every subsequent
|
||
* `send`/`receive`/broadcast cross-check operates under the
|
||
* announced label, no app-side fp ↔ address mapping needed for the
|
||
* receive path.
|
||
*
|
||
* The rename is atomic from a per-peer-mutex perspective — both
|
||
* labels are locked for the duration so concurrent encrypt/decrypt
|
||
* can't observe a half-moved state. Throws if `oldLabel` has no
|
||
* session, or if `newLabel` already does (refuses to overwrite —
|
||
* call `resetSession` first if that's intentional).
|
||
*
|
||
* After alias, the SDK's internal serialization queues
|
||
* (`encryptChains`, `decryptChains`) for `oldLabel` are dropped so
|
||
* future operations don't queue behind a stale chain.
|
||
*
|
||
* V4.8.3 — Prism FR `session-label-asymmetry-v4.8.2.md`.
|
||
*/
|
||
async aliasSession(oldLabel: string, newLabel: string): Promise<void> {
|
||
if (!this.initialized) throw new Error('Not initialized');
|
||
await this.manager.aliasSession(oldLabel, newLabel);
|
||
// The SDK's per-`from` chains are keyed by label; drop the old
|
||
// entries so future `send`/`receive` to either label start with a
|
||
// fresh queue rather than chaining off whatever was last in flight
|
||
// for `oldLabel`.
|
||
this.encryptChains.delete(oldLabel);
|
||
this.decryptChains.delete(oldLabel);
|
||
}
|
||
|
||
/**
|
||
* Accept a peer's rotated identity. Bumps the per-peer identity-version
|
||
* counter so any earlier verification automatically goes stale, then
|
||
* runs the `beforeNewDeviceTrust` gate before the new key is pinned.
|
||
*/
|
||
async acceptIdentityChange(address: string, newIdentityKey: Uint8Array): Promise<void> {
|
||
if (!this.initialized) throw new Error('Not initialized');
|
||
await this.storage.bumpPeerIdentityVersion(address);
|
||
const newFingerprint = await computeFingerprint(
|
||
this.crypto,
|
||
// X3DH stores DH-only "trusted identity"; in this SDK the trusted
|
||
// entry IS the DH public key. We feed it as both args so the
|
||
// fingerprint binds to the rotated key material the user is asked
|
||
// to confirm.
|
||
newIdentityKey,
|
||
newIdentityKey,
|
||
);
|
||
await this.gates.checkNewDeviceTrust(address, newFingerprint);
|
||
await this.manager.acceptIdentityChange(address, newIdentityKey);
|
||
}
|
||
|
||
/** Manually rotate the identity (destructive — see docs) */
|
||
async rotate(): Promise<void> {
|
||
if (!this.initialized) throw new Error('Not initialized');
|
||
|
||
// Rotate locally first
|
||
const newBundle = await this.manager.rotateIdentity();
|
||
|
||
// Rebuild the transport with the new signing key so subsequent
|
||
// signed operations (replenish, delete, register) work
|
||
const identity = await this.storage.getIdentityKeyPair();
|
||
if (!identity) throw new Error('Identity missing after rotate');
|
||
this.transport = new ShadeFetchTransport({
|
||
baseUrl: this.config.prekeyServer,
|
||
crypto: this.crypto,
|
||
signingPrivateKey: identity.signingPrivateKey,
|
||
});
|
||
|
||
// Re-upload the new bundle
|
||
await this.transport.register(
|
||
this.address,
|
||
this.manager.getPublicIdentity(),
|
||
newBundle.signedPreKey,
|
||
[],
|
||
);
|
||
|
||
// Rebuild background tasks so they use the new transport
|
||
if (this._background) {
|
||
this._background.stop();
|
||
this._background = new BackgroundTasks(
|
||
this.manager,
|
||
this.transport,
|
||
this.address,
|
||
this.config,
|
||
);
|
||
this._background.start();
|
||
}
|
||
}
|
||
|
||
/** Manually trigger replenishment (normally background task handles this) */
|
||
async replenish(): Promise<number> {
|
||
if (!this.initialized) throw new Error('Not initialized');
|
||
if (!this._background) return 0;
|
||
return this._background.runReplenish();
|
||
}
|
||
|
||
/**
|
||
* Export an encrypted backup blob that can be restored to a new device.
|
||
*
|
||
* @param passphrase User passphrase (minimum 12 characters)
|
||
* @param knownAddresses Peer addresses whose sessions should be included
|
||
*/
|
||
async exportBackup(passphrase: string, knownAddresses: string[] = []): Promise<string> {
|
||
if (!this.initialized) throw new Error('Not initialized');
|
||
const blob = await exportBackup(this.crypto, this.storage, passphrase, knownAddresses);
|
||
return backupToString(blob);
|
||
}
|
||
|
||
/**
|
||
* Restore state from a backup string. Overwrites existing state.
|
||
* Call this BEFORE initialize() on a fresh device, or after shutdown() + re-init.
|
||
*
|
||
* V3.3: invokes the `beforeBackupImport` gate. The handler receives the
|
||
* fingerprint of the identity *embedded in the backup* — this lets the
|
||
* user OOB-confirm that the backup is theirs before sessions and
|
||
* pinned-trust entries are written to disk.
|
||
*/
|
||
async importBackup(backupString: string, passphrase: string): Promise<void> {
|
||
if (!this.initialized) throw new Error('Not initialized');
|
||
const blob = backupFromString(backupString);
|
||
const payload = await decryptBackup(this.crypto, blob, passphrase);
|
||
const backupFingerprint = await fingerprintFromBackupPayload(this.crypto, payload);
|
||
await this.gates.checkBackupImport(this.address, backupFingerprint);
|
||
await applyBackupPayload(this.storage, payload);
|
||
// Reload identity after restore
|
||
const restored = await this.storage.getIdentityKeyPair();
|
||
if (restored) {
|
||
// Rebuild the manager and transport with the restored identity
|
||
this.manager = new ShadeSessionManager(this.crypto, this.storage, {
|
||
events: this.events,
|
||
...(this.config.observability !== undefined
|
||
? { observability: this.config.observability }
|
||
: {}),
|
||
});
|
||
await this.manager.initialize();
|
||
this.transport = new ShadeFetchTransport({
|
||
baseUrl: this.config.prekeyServer,
|
||
crypto: this.crypto,
|
||
signingPrivateKey: restored.signingPrivateKey,
|
||
});
|
||
}
|
||
}
|
||
|
||
/** Clean shutdown: stop timers, close storage if it supports it */
|
||
async shutdown(): Promise<void> {
|
||
this._background?.stop();
|
||
if (this.transferEngine !== null) this.transferEngine.destroy();
|
||
if (this.controlChannel !== null) this.controlChannel.destroy();
|
||
if (this.webrtcRuntime !== null) {
|
||
this.webrtcRuntime.destroy();
|
||
this.webrtcRuntime = null;
|
||
}
|
||
if (this.workerCrypto !== null) {
|
||
await this.workerCrypto.destroy();
|
||
this.workerCrypto = null;
|
||
}
|
||
// Close storage if it has a close method (SQLite)
|
||
const closable = this.storage as unknown as { close?: () => void | Promise<void> };
|
||
if (typeof closable.close === 'function') {
|
||
await closable.close();
|
||
}
|
||
this.initialized = false;
|
||
}
|
||
|
||
// ─── Worker-Crypto streams (V3.8) ──────────────────────────
|
||
|
||
/**
|
||
* Opt in to Web Workers crypto: subsequent `encryptStream` /
|
||
* `decryptStream` calls offload all AEAD work to a dedicated worker so
|
||
* the main thread stays under the 16 ms-per-frame budget for big
|
||
* uploads. The worker is spawned on first use and self-terminates
|
||
* after `idleTimeoutMs` of inactivity (default 30 s).
|
||
*
|
||
* Bundlers resolve worker URLs differently — the recommended idiom is:
|
||
*
|
||
* ```ts
|
||
* shade.configureWorkerCrypto({
|
||
* workerUrl: new URL('@shade/crypto-web/worker', import.meta.url),
|
||
* });
|
||
* ```
|
||
*
|
||
* See `docs/web-workers.md` for Vite / Webpack / Rollup recipes and
|
||
* Safari notes.
|
||
*/
|
||
configureWorkerCrypto(opts: {
|
||
workerUrl: URL | string;
|
||
idleTimeoutMs?: number;
|
||
}): void {
|
||
this.workerCryptoConfig = opts;
|
||
}
|
||
|
||
/**
|
||
* Encrypt a `ReadableStream<Uint8Array>` of plaintext into stream-chunk
|
||
* wire envelopes via a Web Worker.
|
||
*
|
||
* The caller pre-negotiates `streamId` + `streamSecret` with the peer
|
||
* (typically through `shade.upload()` for HTTP-based delivery, or any
|
||
* other channel). The returned `stream` is a TransformStream:
|
||
* pipe plaintext in, get encrypted chunks out.
|
||
*
|
||
* `laneSha256` resolves once the stream finishes (final chunk emitted
|
||
* with `isLast=true`). Compare it against the receiver's lane sha256
|
||
* for end-to-end integrity proof.
|
||
*
|
||
* Requires `configureWorkerCrypto()` to be called first.
|
||
*/
|
||
encryptStream(
|
||
opts: Omit<CreateEncryptStreamOptions, 'provider'>,
|
||
): Promise<{
|
||
stream: TransformStream<Uint8Array, Uint8Array>;
|
||
laneSha256: Promise<Uint8Array>;
|
||
}> {
|
||
return this.ensureWorkerCrypto().then((provider) =>
|
||
createEncryptStream({ provider, ...opts }),
|
||
);
|
||
}
|
||
|
||
/**
|
||
* Inverse of {@link Shade.encryptStream} — decrypt incoming wire
|
||
* envelopes back into plaintext. Each input chunk MUST be a complete
|
||
* stream-chunk envelope (the wire framing is the caller's job).
|
||
*
|
||
* Requires `configureWorkerCrypto()` to be called first.
|
||
*/
|
||
decryptStream(
|
||
opts: Omit<CreateDecryptStreamOptions, 'provider'>,
|
||
): Promise<{
|
||
stream: TransformStream<Uint8Array, Uint8Array>;
|
||
laneSha256: Promise<Uint8Array>;
|
||
}> {
|
||
return this.ensureWorkerCrypto().then((provider) =>
|
||
createDecryptStream({ provider, ...opts }),
|
||
);
|
||
}
|
||
|
||
/**
|
||
* Direct access to the worker-backed `CryptoProvider`. Use when you
|
||
* want to run a one-off heavy crypto op (X25519 batch DH, big HKDF
|
||
* derivation, etc.) off the main thread without setting up a stream.
|
||
*/
|
||
async getWorkerCrypto(): Promise<WorkerCryptoProvider> {
|
||
return this.ensureWorkerCrypto();
|
||
}
|
||
|
||
private async ensureWorkerCrypto(): Promise<WorkerCryptoProvider> {
|
||
if (this.workerCrypto !== null) return this.workerCrypto;
|
||
if (this.workerCryptoBoot !== null) return this.workerCryptoBoot;
|
||
if (this.workerCryptoConfig === null) {
|
||
throw new Error(
|
||
'Call shade.configureWorkerCrypto({ workerUrl }) before encryptStream()/decryptStream(). See docs/web-workers.md.',
|
||
);
|
||
}
|
||
const cfg = this.workerCryptoConfig;
|
||
this.workerCryptoBoot = (async () => {
|
||
const provider = await createWorkerCryptoProvider({
|
||
workerUrl: cfg.workerUrl,
|
||
...(cfg.idleTimeoutMs !== undefined ? { idleTimeoutMs: cfg.idleTimeoutMs } : {}),
|
||
});
|
||
this.workerCrypto = provider;
|
||
return provider;
|
||
})();
|
||
try {
|
||
return await this.workerCryptoBoot;
|
||
} finally {
|
||
this.workerCryptoBoot = null;
|
||
}
|
||
}
|
||
|
||
// ─── Stream transfers (v0.2.0) ─────────────────────────────
|
||
|
||
/**
|
||
* Configure how stream-transfer chunks reach peers. Provide a resolver
|
||
* that returns the peer's HTTP base URL (e.g. by looking up a
|
||
* `transfer.baseUrl` field in your prekey-bundle metadata or a static
|
||
* directory map). If unset, `upload()` rejects with a clear error.
|
||
*
|
||
* Optionally also override the control-envelope transport (defaults to
|
||
* HTTP POSTs to `<base>/v1/transfer/control`).
|
||
*/
|
||
configureTransfers(opts: {
|
||
/**
|
||
* Resolver for the peer's HTTP base URL (used by the default
|
||
* `ShadeTransferHttpTransport` to POST chunks). Optional when a
|
||
* custom `transport` and `envelopeTransport` are supplied — e.g.
|
||
* for pull-mode browser servers (`@shade/files transferQueueRoute`)
|
||
* which never POST chunks anywhere.
|
||
*/
|
||
resolveBaseUrl?: (peerAddress: string) => Promise<string>;
|
||
/**
|
||
* Override the chunk-level transport. Defaults to
|
||
* `ShadeTransferHttpTransport` (HTTP POSTs per chunk) when
|
||
* `resolveBaseUrl` is supplied. Required when `resolveBaseUrl`
|
||
* is omitted.
|
||
*/
|
||
transport?: ITransferTransport;
|
||
/**
|
||
* Override the control-envelope transport. Defaults to HTTP POSTs
|
||
* to `<base>/v1/transfer/control` when `resolveBaseUrl` is
|
||
* supplied. Required when `resolveBaseUrl` is omitted.
|
||
*/
|
||
envelopeTransport?: ControlEnvelopeTransport;
|
||
}): void {
|
||
if (opts.resolveBaseUrl === undefined) {
|
||
if (opts.transport === undefined || opts.envelopeTransport === undefined) {
|
||
throw new Error(
|
||
'configureTransfers: resolveBaseUrl is required unless both `transport` and `envelopeTransport` are supplied (e.g. for pull-mode queue servers).',
|
||
);
|
||
}
|
||
this.peerBaseUrlResolver = async () => {
|
||
throw new Error(
|
||
'resolveBaseUrl was not configured — this Shade is in queue/pull mode and does not POST chunks. Configure a custom transport instead.',
|
||
);
|
||
};
|
||
} else {
|
||
this.peerBaseUrlResolver = opts.resolveBaseUrl;
|
||
}
|
||
this.transferTransportOverride = opts.transport ?? null;
|
||
if (opts.envelopeTransport !== undefined) {
|
||
this.envelopeOutboxes = opts.envelopeTransport;
|
||
} else if (opts.resolveBaseUrl !== undefined) {
|
||
this.envelopeOutboxes = new HttpEnvelopeTransport(opts.resolveBaseUrl, this.address);
|
||
} else {
|
||
throw new Error(
|
||
'configureTransfers: envelopeTransport is required when resolveBaseUrl is omitted.',
|
||
);
|
||
}
|
||
}
|
||
|
||
/**
|
||
* Deliver a freshly-encrypted ratchet envelope to a peer using the
|
||
* configured envelope transport (HTTP POST to `/v1/transfer/control` by
|
||
* default). Used by `@shade/files` for RPC plaintext delivery.
|
||
*/
|
||
async deliverControlEnvelope(peerAddress: string, envelope: ShadeEnvelope): Promise<void> {
|
||
if (this.envelopeOutboxes === null) {
|
||
throw new Error(
|
||
'Call shade.configureTransfers({ resolveBaseUrl }) before deliverControlEnvelope()',
|
||
);
|
||
}
|
||
await this.envelopeOutboxes.send(peerAddress, envelope);
|
||
}
|
||
|
||
/**
|
||
* Upload bytes to a peer. Returns a `TransferHandle` that can be paused/
|
||
* aborted and awaited. Requires `configureTransfers` to be called first.
|
||
*
|
||
* V3.3: when the file size is at or above the configured threshold
|
||
* (default 10 MiB) and the peer is not yet verified, the registered
|
||
* `beforeFirstLargeFile` handler is invoked. Rejection throws
|
||
* `FingerprintNotVerifiedError` before any bytes hit the wire.
|
||
*
|
||
* V3.9: pass `thumbnail: { bytes, mime }` to attach a pre-generated
|
||
* preview, or `generateThumbnail: true` to auto-derive a 256x256 preview
|
||
* from an image input in browser-class runtimes (no-op elsewhere). The
|
||
* thumbnail is shipped as a *separate* E2EE stream and referenced from
|
||
* the main stream's `fileMetadata`.
|
||
*/
|
||
async upload(opts: ShadeUploadOptions): Promise<TransferHandle> {
|
||
if (!this.initialized) throw new Error('Not initialized');
|
||
const size = inferTransferSize(opts);
|
||
if (size !== null && size >= this.gates.getFirstLargeFileThreshold()) {
|
||
// Establish the session up-front so we have a fingerprint to gate on.
|
||
// For peers we've never contacted, this is the TOFU moment where the
|
||
// gate matters most.
|
||
if ((await this.storage.getSession(opts.to)) === null) {
|
||
await this.ensureSession(opts.to);
|
||
}
|
||
const fingerprint = await this.manager.getRemoteFingerprint(opts.to);
|
||
await this.gates.checkFirstLargeFile(opts.to, fingerprint, size);
|
||
}
|
||
const engine = await this.engine();
|
||
const thumbnail = await this.resolveThumbnail(opts);
|
||
if (thumbnail !== null) {
|
||
const fileMeta: StreamFileMetadata = {
|
||
...(opts.metadata?.fileMetadata ?? {}),
|
||
thumbnailStreamId: thumbnail.streamId,
|
||
thumbnailHash: thumbnail.hashB64,
|
||
thumbnailMime: thumbnail.mime,
|
||
thumbnailBytes: thumbnail.bytes,
|
||
};
|
||
const merged: TransferOptions = {
|
||
...opts,
|
||
metadata: {
|
||
...(opts.metadata ?? {}),
|
||
fileMetadata: fileMeta,
|
||
},
|
||
};
|
||
return engine.upload(merged);
|
||
}
|
||
return engine.upload(opts);
|
||
}
|
||
|
||
/**
|
||
* Coordinate the thumbnail-side of a V3.9 upload. Resolves to either
|
||
* - `null` — no thumbnail will be attached (caller passed neither
|
||
* `thumbnail` nor a generator that produced bytes), or
|
||
* - the streamId + sha256 + mime + bytes of the thumbnail-stream that
|
||
* has now been kicked off (it runs to completion in the background;
|
||
* the main upload's `done()` is independent).
|
||
*/
|
||
private async resolveThumbnail(opts: ShadeUploadOptions): Promise<{
|
||
streamId: string;
|
||
hashB64: string;
|
||
mime: ThumbnailMime;
|
||
bytes: number;
|
||
} | null> {
|
||
let bytes: Uint8Array | null = null;
|
||
let mime: ThumbnailMime | null = null;
|
||
if (opts.thumbnail !== undefined) {
|
||
bytes = opts.thumbnail.bytes;
|
||
mime = opts.thumbnail.mime;
|
||
} else if (opts.generateThumbnail !== undefined && opts.generateThumbnail !== false) {
|
||
const genOpts: ThumbnailGenerationOptions =
|
||
opts.generateThumbnail === true ? {} : opts.generateThumbnail;
|
||
const gen = await generateThumbnailFromBlob(opts.input, genOpts);
|
||
if (gen !== null) {
|
||
bytes = gen.bytes;
|
||
mime = gen.mime;
|
||
}
|
||
}
|
||
if (bytes === null || mime === null) return null;
|
||
if (bytes.byteLength > THUMBNAIL_MAX_BYTES) {
|
||
throw new Error(
|
||
`thumbnail size ${bytes.byteLength} exceeds THUMBNAIL_MAX_BYTES (${THUMBNAIL_MAX_BYTES})`,
|
||
);
|
||
}
|
||
if (!isAllowedThumbnailMime(mime)) {
|
||
throw new Error(`thumbnail mime ${mime} not in allowlist`);
|
||
}
|
||
const hash = sha256Once(bytes);
|
||
const hashB64 = bytesToBase64Std(hash);
|
||
const engine = await this.engine();
|
||
// Ship the thumbnail FIRST so the receiver can present a preview the
|
||
// moment the main `stream-init` references it. Single lane, single
|
||
// chunk — at ≤ 64 KiB the parallelism overhead would dominate.
|
||
const handle = await engine.upload({
|
||
to: opts.to,
|
||
input: bytes,
|
||
lanes: 1,
|
||
chunkSize: Math.max(1, bytes.byteLength),
|
||
metadata: {
|
||
contentType: mime,
|
||
userMetadata: {
|
||
shadeThumbnail: '1',
|
||
},
|
||
},
|
||
});
|
||
// Don't await `done()` — the main upload should not block on the
|
||
// thumbnail finishing. Errors on the preview are surfaced via the
|
||
// returned handle's events (consumer can listen if they care).
|
||
handle.done().catch((err) => {
|
||
console.warn('[Shade] thumbnail transfer failed:', err);
|
||
});
|
||
return {
|
||
streamId: handle.streamId,
|
||
hashB64,
|
||
mime,
|
||
bytes: bytes.byteLength,
|
||
};
|
||
}
|
||
|
||
/**
|
||
* Subscribe to incoming transfers from peers. Handler is invoked when a
|
||
* `stream-init` arrives; the handler MUST call `incoming.accept(...)` to
|
||
* begin receiving (or `incoming.decline(...)` to reject).
|
||
*/
|
||
async onIncomingTransfer(
|
||
handler: (incoming: IncomingTransfer) => void | Promise<void>,
|
||
): Promise<() => void> {
|
||
return (await this.engine()).onIncomingTransfer(handler);
|
||
}
|
||
|
||
/**
|
||
* Mount the **pull-mode** transfer routes on a Hono app. Mount under
|
||
* any base path: `app.route('/api/v1/shade-files', shade.transferQueueRoute())`.
|
||
*
|
||
* Configures this Shade instance to queue all outbound chunks +
|
||
* control envelopes per peer instead of POSTing them. Browser-style
|
||
* receivers drain the queue via long-polling — no inbound HTTP
|
||
* listener required on the receiver.
|
||
*
|
||
* Routes mounted (relative to the base path):
|
||
* POST /queue — long-poll the per-peer outbound queue
|
||
* POST /v1/transfer/:streamId/chunk — receive incoming chunks (browser → server)
|
||
* GET /v1/transfer/:streamId/state — resume-state lookup
|
||
* POST /v1/transfer/control — receive incoming control envelopes
|
||
* GET /v1/transfer/health — peer reachability probe
|
||
*
|
||
* **Idempotent**: calling twice returns a fresh `Hono` app each
|
||
* time but reuses the underlying queue + transport (so the engine
|
||
* stays single).
|
||
*
|
||
* **Ordering**: must be called **before** `shade.files.serve(...)`
|
||
* (or any other path that builds the engine), because configuring
|
||
* the queue transport mutates the transfer stack. Calling after the
|
||
* engine is built throws.
|
||
*/
|
||
async transferQueueRoute(opts: TransferQueueRouteOptions = {}): Promise<Hono> {
|
||
if (this.transferEngine !== null && this.transferTransportOverride === null) {
|
||
throw new Error(
|
||
'transferQueueRoute(): the transfer engine has already been built with the default HTTP transport. Call transferQueueRoute() before any upload()/onIncomingTransfer()/configureTransfers().',
|
||
);
|
||
}
|
||
const { OutboundQueue, QueueTransferTransport } = await import('@shade/transfer');
|
||
if (this.transferQueue === null) {
|
||
this.transferQueue = new OutboundQueue({
|
||
...(opts.maxEventsPerPeer !== undefined ? { maxEventsPerPeer: opts.maxEventsPerPeer } : {}),
|
||
...(opts.idleEvictionMs !== undefined ? { idleEvictionMs: opts.idleEvictionMs } : {}),
|
||
});
|
||
}
|
||
if (this.transferTransportOverride === null) {
|
||
const queueTransport = new QueueTransferTransport(this.transferQueue);
|
||
const queueEnvelopeTransport = new QueueEnvelopeTransport(this.transferQueue);
|
||
this.configureTransfers({
|
||
transport: queueTransport,
|
||
envelopeTransport: queueEnvelopeTransport,
|
||
});
|
||
}
|
||
const queue = this.transferQueue;
|
||
const blockMs = opts.blockMs ?? 30_000;
|
||
const maxBlockMs = opts.maxBlockMs ?? 55_000;
|
||
const engine = await this.engine();
|
||
const { createTransferRoutes, PermissiveAuthenticator } = await import('@shade/transfer');
|
||
const app = await createTransferRoutes(engine, {
|
||
authenticator: PermissiveAuthenticator,
|
||
});
|
||
app.post('/v1/transfer/control', async (c) => {
|
||
const senderAddress = c.req.header('X-Shade-Sender-Address');
|
||
if (senderAddress === undefined || senderAddress === '') {
|
||
return c.json({ error: 'missing X-Shade-Sender-Address' }, 400);
|
||
}
|
||
const ab = await c.req.arrayBuffer();
|
||
const bytes = new Uint8Array(ab);
|
||
try {
|
||
await this.acceptTransferEnvelope(senderAddress, bytes);
|
||
} catch (err) {
|
||
return c.json({ error: (err as Error).message }, 400);
|
||
}
|
||
return c.json({ ok: true });
|
||
});
|
||
// Long-poll endpoint.
|
||
app.post('/queue', async (c) => {
|
||
const senderAddress = c.req.header('X-Shade-Sender-Address');
|
||
if (senderAddress === undefined || senderAddress === '') {
|
||
return c.json({ error: 'missing X-Shade-Sender-Address' }, 400);
|
||
}
|
||
let body: { since?: unknown; blockMs?: unknown };
|
||
try {
|
||
body = (await c.req.json()) as { since?: unknown; blockMs?: unknown };
|
||
} catch {
|
||
return c.json({ error: 'invalid JSON body' }, 400);
|
||
}
|
||
const since = typeof body.since === 'number' && Number.isFinite(body.since) ? body.since : 0;
|
||
const requestedBlockMs =
|
||
typeof body.blockMs === 'number' && Number.isFinite(body.blockMs)
|
||
? Math.max(0, Math.min(maxBlockMs, body.blockMs))
|
||
: blockMs;
|
||
// Bun-side short-circuit if the request was aborted while we
|
||
// were holding the long-poll. AbortSignal from the request body
|
||
// is already surfaced via `c.req.raw.signal` in Hono.
|
||
const events = await queue.drain(senderAddress, since, requestedBlockMs, c.req.raw.signal);
|
||
return c.json({
|
||
events: events.map((e) => ({
|
||
id: e.id,
|
||
timestampMs: e.timestampMs,
|
||
kind: e.kind,
|
||
bytesB64: bytesToBase64Std(e.bytes),
|
||
...(e.kind === 'chunk' ? { meta: e.meta } : {}),
|
||
})),
|
||
nextSince: events.length > 0 ? events[events.length - 1]!.id : since,
|
||
});
|
||
});
|
||
return app;
|
||
}
|
||
|
||
/**
|
||
* Mount the receiver-side HTTP routes on a Hono app. Mount under any
|
||
* base path: `app.route('/shade', await shade.transferRoute())`.
|
||
*
|
||
* Routes:
|
||
* POST /v1/transfer/:streamId/chunk — wire-encoded 0x11 chunks
|
||
* GET /v1/transfer/:streamId/state — resume-state lookup
|
||
* POST /v1/transfer/control — wire-encoded 0x02 control envelopes
|
||
* GET /v1/transfer/health — peer reachability probe
|
||
*/
|
||
async transferRoute(): Promise<Hono> {
|
||
const engine = await this.engine();
|
||
const { createTransferRoutes, PermissiveAuthenticator } = await import('@shade/transfer');
|
||
const app = await createTransferRoutes(engine, {
|
||
authenticator: PermissiveAuthenticator,
|
||
});
|
||
// Add the control-envelope POST route on top.
|
||
app.post('/v1/transfer/control', async (c) => {
|
||
const senderAddress = c.req.header('X-Shade-Sender-Address');
|
||
if (senderAddress === undefined || senderAddress === '') {
|
||
return c.json({ error: 'missing X-Shade-Sender-Address' }, 400);
|
||
}
|
||
const ab = await c.req.arrayBuffer();
|
||
const bytes = new Uint8Array(ab);
|
||
try {
|
||
await this.acceptTransferEnvelope(senderAddress, bytes);
|
||
} catch (err) {
|
||
return c.json({ error: (err as Error).message }, 400);
|
||
}
|
||
return c.json({ ok: true });
|
||
});
|
||
return app;
|
||
}
|
||
|
||
/**
|
||
* Low-level entry for custom transports: hand a `0x02` ratchet envelope
|
||
* (control-plane) or a `0x11` stream-chunk envelope to the engine.
|
||
* Used internally by `transferRoute()`.
|
||
*/
|
||
async acceptTransferEnvelope(from: string, env: ShadeEnvelope | Uint8Array): Promise<void> {
|
||
const engine = await this.engine();
|
||
if (env instanceof Uint8Array) {
|
||
const kind = inspectEnvelopeType(env);
|
||
if (kind === 'stream-chunk') {
|
||
// Engine extracts laneId/seq from the wire bytes via decodeStreamChunk.
|
||
const headers = parseChunkHeader(env);
|
||
await engine.receiveChunk(from, headers.streamId, headers.laneId, headers.seq, env);
|
||
return;
|
||
}
|
||
if (kind === 'ratchet' || kind === 'prekey') {
|
||
const decoded = decodeEnvelope(env);
|
||
await this.controlChannel!.acceptEnvelope(from, decoded);
|
||
return;
|
||
}
|
||
throw new Error(`Unknown envelope type ${kind}`);
|
||
}
|
||
// Already-decoded envelope (ratchet or prekey)
|
||
await this.controlChannel!.acceptEnvelope(from, env);
|
||
}
|
||
|
||
// ─── V3.11 WebRTC P2P transport ────────────────────────────
|
||
|
||
/**
|
||
* Opt in to direct peer-to-peer chunk delivery via WebRTC.
|
||
*
|
||
* When configured, `upload()` builds a `[WebRTC, HTTP]`
|
||
* {@link MultiTransportFallback}: P2P first, HTTP as automatic
|
||
* fallback. Signaling (SDP offer/answer + trickle-ICE) rides on top
|
||
* of `Shade.send` / `Shade.onMessage` — no out-of-band server needed.
|
||
*
|
||
* Must be called BEFORE the first `upload()` / `onIncomingTransfer()`
|
||
* (those instantiate the transfer engine, which captures the
|
||
* transport stack at construction time). Calling later throws.
|
||
*
|
||
* The `factory` argument is the WebRTC adapter — `nativeRtcFactory()`
|
||
* for browsers, a custom one for Node-class environments
|
||
* (`node-datachannel`, `wrtc`, etc.). Set `iceServers` to override the
|
||
* default public STUN list, or supply TURN credentials for paranoid
|
||
* NATs:
|
||
*
|
||
* ```ts
|
||
* import { nativeRtcFactory } from '@shade/transport-webrtc';
|
||
* shade.configureWebRTC({
|
||
* factory: nativeRtcFactory(),
|
||
* iceServers: [
|
||
* { urls: 'stun:stun.l.google.com:19302' },
|
||
* { urls: 'turn:turn.example.com:3478', username: 'u', credential: 'p' },
|
||
* ],
|
||
* });
|
||
* ```
|
||
*/
|
||
configureWebRTC(opts: ShadeWebRtcConfig): void {
|
||
if (this.transferEngine !== null) {
|
||
throw new Error(
|
||
'shade.configureWebRTC() must be called before upload()/onIncomingTransfer() builds the engine',
|
||
);
|
||
}
|
||
this.webrtcConfig = opts;
|
||
}
|
||
|
||
/**
|
||
* Returns the live WebRTC runtime (signaling channel + connection
|
||
* manager + transport) if `configureWebRTC` was called and `engine()`
|
||
* has been instantiated. Useful for diagnostics: peek
|
||
* `runtime.manager.isConnected('alice')` to see whether a P2P link is
|
||
* live, or wire `runtime.fallback.onSwitch(...)` to log demotions.
|
||
*/
|
||
getWebRtcRuntime(): ShadeWebRtcRuntime | null {
|
||
return this.webrtcRuntime;
|
||
}
|
||
|
||
// ─── Internals ─────────────────────────────────────────────
|
||
|
||
private async engine(): Promise<TransferEngine> {
|
||
if (this.transferEngine !== null) return this.transferEngine;
|
||
if (this.peerBaseUrlResolver === null || this.envelopeOutboxes === null) {
|
||
throw new Error(
|
||
'Call shade.configureTransfers({ resolveBaseUrl }) before using upload()/onIncomingTransfer()',
|
||
);
|
||
}
|
||
this.controlChannel = new ShadeControlChannel(this, this.envelopeOutboxes);
|
||
let transport: ITransferTransport;
|
||
let webrtcRuntime: ShadeWebRtcRuntime | null = null;
|
||
if (this.transferTransportOverride !== null) {
|
||
// Custom transport (queue, in-memory, custom adapter) — used as-is.
|
||
// WebRTC fallback only attaches when the default HTTP transport is
|
||
// active because WebRTC's `MultiTransportFallback` is HTTP-shaped.
|
||
transport = this.transferTransportOverride;
|
||
} else {
|
||
const httpTransport: ITransferTransport = new ShadeTransferHttpTransport({
|
||
resolveBaseUrl: this.peerBaseUrlResolver,
|
||
authenticator: await this.makeAuthenticator(),
|
||
});
|
||
transport = httpTransport;
|
||
if (this.webrtcConfig !== null) {
|
||
webrtcRuntime = await this.buildWebRtcRuntime(this.webrtcConfig, httpTransport);
|
||
transport = webrtcRuntime.fallback;
|
||
}
|
||
}
|
||
|
||
this.transferEngine = new TransferEngine({
|
||
crypto: this.crypto,
|
||
controlChannel: this.controlChannel,
|
||
transport,
|
||
myAddress: this.address,
|
||
...(this.config.observability !== undefined
|
||
? { observability: this.config.observability }
|
||
: {}),
|
||
});
|
||
if (webrtcRuntime !== null) {
|
||
// Receiver-hooks need to dispatch into the freshly-built engine.
|
||
webrtcRuntime.attachEngine(this.transferEngine);
|
||
this.webrtcRuntime = webrtcRuntime;
|
||
}
|
||
return this.transferEngine;
|
||
}
|
||
|
||
/**
|
||
* Dynamically import `@shade/transport-webrtc`, wire its signaling
|
||
* channel onto our `Shade.send`/`Shade.onMessage`, and build a
|
||
* MultiTransportFallback that prefers WebRTC then falls back to HTTP.
|
||
*/
|
||
private async buildWebRtcRuntime(
|
||
cfg: ShadeWebRtcConfig,
|
||
httpTransport: ITransferTransport,
|
||
): Promise<ShadeWebRtcRuntime> {
|
||
// `@shade/transport-webrtc` is an optional peer dep — keep the
|
||
// import lazy so consumers that don't use WebRTC don't pay for it.
|
||
const moduleId = '@shade/transport-webrtc';
|
||
const mod = (await import(moduleId)) as typeof import('@shade/transport-webrtc');
|
||
const {
|
||
WebRtcSignalingChannel,
|
||
WebRtcConnectionManager,
|
||
WebRtcTransferTransport,
|
||
createShadeBridgeFromShade,
|
||
} = mod;
|
||
|
||
const signaling = new WebRtcSignalingChannel(createShadeBridgeFromShade(this));
|
||
let engineRef: TransferEngine | null = null;
|
||
const manager = new WebRtcConnectionManager({
|
||
factory: cfg.factory,
|
||
signaling,
|
||
...(cfg.iceServers !== undefined ||
|
||
cfg.iceTransportPolicy !== undefined ||
|
||
cfg.bundlePolicy !== undefined
|
||
? {
|
||
config: {
|
||
...(cfg.iceServers !== undefined ? { iceServers: cfg.iceServers } : {}),
|
||
...(cfg.iceTransportPolicy !== undefined
|
||
? { iceTransportPolicy: cfg.iceTransportPolicy }
|
||
: {}),
|
||
...(cfg.bundlePolicy !== undefined ? { bundlePolicy: cfg.bundlePolicy } : {}),
|
||
},
|
||
}
|
||
: {}),
|
||
...(cfg.connectTimeoutMs !== undefined ? { connectTimeoutMs: cfg.connectTimeoutMs } : {}),
|
||
receiver: {
|
||
onChunk: async (from, streamId, laneId, seq, envelope) => {
|
||
if (engineRef === null) {
|
||
throw new Error('webrtc receiver hook fired before engine attached');
|
||
}
|
||
return engineRef.receiveChunk(from, streamId, laneId, seq, envelope);
|
||
},
|
||
onResumeQuery: async (from, streamId) => {
|
||
if (engineRef === null) return null;
|
||
return engineRef.getResumeState(from, streamId);
|
||
},
|
||
},
|
||
});
|
||
const webrtcTransport = new WebRtcTransferTransport({
|
||
manager,
|
||
...(cfg.requestTimeoutMs !== undefined ? { requestTimeoutMs: cfg.requestTimeoutMs } : {}),
|
||
...(cfg.backpressureThresholdBytes !== undefined
|
||
? { backpressureThresholdBytes: cfg.backpressureThresholdBytes }
|
||
: {}),
|
||
});
|
||
const fallback = new MultiTransportFallback([
|
||
{ name: 'webrtc', transport: webrtcTransport },
|
||
{ name: 'http', transport: httpTransport },
|
||
]);
|
||
|
||
return {
|
||
signaling,
|
||
manager,
|
||
transport: webrtcTransport,
|
||
fallback,
|
||
attachEngine(engine) {
|
||
engineRef = engine;
|
||
},
|
||
destroy() {
|
||
manager.destroy();
|
||
signaling.destroy();
|
||
},
|
||
};
|
||
}
|
||
|
||
private async makeAuthenticator(): Promise<ShadeTransferAuthenticator> {
|
||
const identity = await this.storage.getIdentityKeyPair();
|
||
if (identity === null) throw new Error('Identity not initialized');
|
||
return new ShadeTransferAuthenticator(this.crypto, this.address, identity.signingPrivateKey);
|
||
}
|
||
|
||
/** Returns a list of in-flight stream transfers from storage (resume support). */
|
||
async listTransfers(filter?: {
|
||
direction?: 'send' | 'receive';
|
||
}): Promise<TransferSummary[]> {
|
||
if (this.storage.listActiveStreamStates === undefined) return [];
|
||
const rows = await this.storage.listActiveStreamStates(filter?.direction);
|
||
return rows.map((s) => ({
|
||
streamId: s.streamId,
|
||
direction: s.direction,
|
||
peerAddress: s.peerAddress,
|
||
status: s.status,
|
||
bytesProcessed: 0, // computed from laneState
|
||
createdAt: s.createdAt,
|
||
updatedAt: s.updatedAt,
|
||
metadata: tryParseMetadata(s.metadataJson),
|
||
}));
|
||
}
|
||
|
||
/**
|
||
* Drop persisted stream-state records whose `updatedAt` is strictly
|
||
* less than `olderThan` (Unix ms). Idempotent. Returns silently when
|
||
* the configured storage backend does not implement stream-state
|
||
* persistence (e.g. memory storage in tests).
|
||
*
|
||
* Recommended usage: schedule on a daily cron with a 14-day horizon
|
||
* — see `docs/streams.md` § Retention. The `bun-server` SDK template
|
||
* wires this up by default.
|
||
*/
|
||
async pruneStreamStates(olderThan: number): Promise<void> {
|
||
if (this.storage.pruneStreamStates === undefined) return;
|
||
await this.storage.pruneStreamStates(olderThan);
|
||
}
|
||
|
||
private async ensureSession(address: string): Promise<void> {
|
||
// Deduplicate concurrent establishment requests
|
||
const existing = this.establishing.get(address);
|
||
if (existing) {
|
||
await existing;
|
||
return;
|
||
}
|
||
|
||
const promise = (async () => {
|
||
const bundle = await this.transport.fetchBundle(address);
|
||
await this.manager.initSessionFromBundle(address, bundle);
|
||
})();
|
||
|
||
this.establishing.set(address, promise);
|
||
try {
|
||
await promise;
|
||
} finally {
|
||
this.establishing.delete(address);
|
||
}
|
||
}
|
||
}
|
||
|
||
function bytesToBase64Std(bytes: Uint8Array): string {
|
||
let bin = '';
|
||
for (let i = 0; i < bytes.length; i++) bin += String.fromCharCode(bytes[i]!);
|
||
return btoa(bin);
|
||
}
|
||
|
||
function tryParseMetadata(json: string): import('@shade/streams').StreamMetadata | null {
|
||
try {
|
||
return JSON.parse(json) as import('@shade/streams').StreamMetadata;
|
||
} catch {
|
||
return null;
|
||
}
|
||
}
|
||
|
||
/**
|
||
* Best-effort plaintext-size inference for a `TransferOptions.input`.
|
||
* Returns null when the size is genuinely unknowable (raw `ReadableStream`
|
||
* without a metadata hint), so the caller can decide whether to gate.
|
||
*/
|
||
function inferTransferSize(opts: TransferOptions): number | null {
|
||
if (typeof opts.metadata?.sizeBytes === 'number') return opts.metadata.sizeBytes;
|
||
const input = opts.input;
|
||
if (input instanceof Uint8Array) return input.byteLength;
|
||
// Blob and File both expose `.size`. Use a structural check so we don't
|
||
// depend on lib.dom typings inside the SDK build.
|
||
if (typeof (input as unknown as { size?: unknown }).size === 'number') {
|
||
return (input as unknown as { size: number }).size;
|
||
}
|
||
return null;
|
||
}
|
||
|
||
/**
|
||
* Compute the safety-number fingerprint for the identity embedded in a
|
||
* decrypted backup payload. Used by `Shade.importBackup` to drive the
|
||
* `beforeBackupImport` gate before any state is overwritten.
|
||
*/
|
||
async function fingerprintFromBackupPayload(
|
||
crypto: SubtleCryptoProvider,
|
||
payload: import('./backup.js').BackupPayload,
|
||
): Promise<string> {
|
||
if (payload.identity === null) {
|
||
// No identity in the backup means there's nothing to fingerprint.
|
||
// Return a stable sentinel so the gate handler can still display
|
||
// something meaningful instead of throwing here.
|
||
return 'no-identity-in-backup';
|
||
}
|
||
const id = deserializeIdentityKeyPair(payload.identity);
|
||
return computeFingerprint(crypto, id.signingPublicKey, id.dhPublicKey);
|
||
}
|
||
|
||
function parseChunkHeader(bytes: Uint8Array): {
|
||
streamId: string;
|
||
laneId: number;
|
||
seq: bigint;
|
||
} {
|
||
// [0]=ver [1]=type [2..18]=streamId(16) [18..22]=laneId u32 [22..30]=seq u64
|
||
if (bytes.length < 30) throw new Error('truncated stream-chunk header');
|
||
const view = new DataView(bytes.buffer, bytes.byteOffset);
|
||
const sidBytes = bytes.slice(2, 18);
|
||
const laneId = view.getUint32(18, false);
|
||
const seq = view.getBigUint64(22, false);
|
||
// Encode streamId as base64url
|
||
let bin = '';
|
||
for (let i = 0; i < sidBytes.length; i++) bin += String.fromCharCode(sidBytes[i]!);
|
||
const streamId = btoa(bin).replace(/\+/g, '-').replace(/\//g, '_').replace(/=+$/, '');
|
||
return { streamId, laneId, seq };
|
||
}
|
||
|
||
// ─── Queue-mode (pull) envelope transport ─────────────────────
|
||
|
||
/**
|
||
* Configuration for {@link Shade.transferQueueRoute}. All fields are
|
||
* optional with sensible production defaults.
|
||
*/
|
||
export interface TransferQueueRouteOptions {
|
||
/**
|
||
* Long-poll timeout in milliseconds. Server holds the request open
|
||
* up to this long before returning an empty `events` array. Default
|
||
* 30_000.
|
||
*/
|
||
blockMs?: number;
|
||
/**
|
||
* Hard cap on long-poll timeout (clamps client-supplied `blockMs`).
|
||
* Default 55_000 — under typical reverse-proxy idle thresholds (60s
|
||
* on most CDNs).
|
||
*/
|
||
maxBlockMs?: number;
|
||
/**
|
||
* Per-peer ring-buffer size. When the queue is full, oldest events
|
||
* are dropped on enqueue. Receivers detect the gap via missing
|
||
* sequence numbers and re-resume from `since=0`. Default 1000.
|
||
*/
|
||
maxEventsPerPeer?: number;
|
||
/**
|
||
* Drop a peer's queue + reject pending pollers after this much
|
||
* silence. Default 10 minutes. Setting to `0` disables idle-eviction.
|
||
*/
|
||
idleEvictionMs?: number;
|
||
}
|
||
|
||
/**
|
||
* `ControlEnvelopeTransport` that enqueues outbound envelopes into an
|
||
* `OutboundQueue` for browser-style receivers to long-poll. Mirrors
|
||
* `HttpEnvelopeTransport` shape (one `send(peer, envelope)` method);
|
||
* the difference is the destination — local queue, not remote HTTP.
|
||
*/
|
||
class QueueEnvelopeTransport implements ControlEnvelopeTransport {
|
||
constructor(private readonly queue: OutboundQueueLike) {}
|
||
async send(peerAddress: string, envelope: ShadeEnvelope): Promise<void> {
|
||
const bytes = encodeEnvelope(envelope);
|
||
const event: QueuedEventInput = { kind: 'envelope', bytes };
|
||
this.queue.enqueue(peerAddress, event);
|
||
}
|
||
}
|
||
|
||
// ─── Default HTTP envelope transport ──────────────────────────
|
||
|
||
class HttpEnvelopeTransport implements ControlEnvelopeTransport {
|
||
constructor(
|
||
private readonly resolveBaseUrl: (peerAddress: string) => Promise<string>,
|
||
private readonly myAddress: string,
|
||
) {}
|
||
async send(peerAddress: string, envelope: ShadeEnvelope): Promise<void> {
|
||
const base = (await this.resolveBaseUrl(peerAddress)).replace(/\/$/, '');
|
||
const url = `${base}/v1/transfer/control`;
|
||
const bytes = encodeEnvelope(envelope);
|
||
const res = await fetch(url, {
|
||
method: 'POST',
|
||
headers: {
|
||
'Content-Type': 'application/octet-stream',
|
||
'X-Shade-Sender-Address': this.myAddress,
|
||
},
|
||
body: bytes as unknown as never,
|
||
});
|
||
if (!res.ok) {
|
||
throw new Error(`control envelope POST failed: ${res.status} ${await res.text()}`);
|
||
}
|
||
}
|
||
}
|
||
|
||
// ─── Helpers ─────────────────────────────────────────────────
|
||
|
||
async function resolveStorage(spec: StorageSpec): Promise<StorageProvider> {
|
||
if (typeof spec === 'object' && 'getIdentityKeyPair' in spec) {
|
||
return spec;
|
||
}
|
||
|
||
if (spec === 'memory') {
|
||
return new MemoryStorage();
|
||
}
|
||
|
||
if (typeof spec === 'string' && spec.startsWith('sqlite:')) {
|
||
const path = spec.slice('sqlite:'.length);
|
||
const { SQLiteStorage } = await import('@shade/storage-sqlite');
|
||
return new SQLiteStorage(path);
|
||
}
|
||
|
||
if (typeof spec === 'object' && spec.type === 'postgres') {
|
||
// Dynamic import keeps @shade/storage-postgres optional — consumers that
|
||
// never use postgres don't need to install it. The string-form import
|
||
// path makes the resolver lazy at type-check time too.
|
||
const moduleId = '@shade/storage-postgres';
|
||
const mod = (await import(moduleId)) as {
|
||
PostgresStorage: { create(url: string): Promise<StorageProvider> };
|
||
};
|
||
return mod.PostgresStorage.create(spec.url);
|
||
}
|
||
|
||
if (typeof spec === 'object' && spec.type === 'indexeddb') {
|
||
// Dynamic import keeps @shade/storage-indexeddb optional — Node-only
|
||
// consumers don't need to install a browser-only adapter.
|
||
const moduleId = '@shade/storage-indexeddb';
|
||
const mod = (await import(moduleId)) as {
|
||
IndexedDBStorage: { create(opts: { dbName?: string }): Promise<StorageProvider> };
|
||
};
|
||
const opts: { dbName?: string } = {};
|
||
if (spec.dbName !== undefined) opts.dbName = spec.dbName;
|
||
return mod.IndexedDBStorage.create(opts);
|
||
}
|
||
|
||
throw new Error(`Unsupported storage spec: ${JSON.stringify(spec)}`);
|
||
}
|
||
|
||
async function resolveAddress(storage: StorageProvider): Promise<string> {
|
||
// Try to load a persisted address, else generate a random one and save it.
|
||
// We reuse the config table by storing a special key.
|
||
// Since StorageProvider doesn't expose a generic key-value, we just use
|
||
// the local registration ID as a deterministic fallback.
|
||
const id = await storage.getLocalRegistrationId();
|
||
return `device:${id}`;
|
||
}
|
||
|
||
function normalize(fp: string): string {
|
||
return fp.replace(/\s+/g, ' ').trim();
|
||
}
|