265 lines
9.4 KiB
TypeScript
265 lines
9.4 KiB
TypeScript
|
|
// NOTE: Replace this with libauth sha256
|
|
import { toExtendedJson } from '@xo-cash/utils';
|
|
import { z } from 'zod';
|
|
|
|
import { HTTP_STATUS_CODE_NOT_ACCEPTED } from '../constants.ts';
|
|
|
|
import type{ AuthSecp256k1 } from '../auth/auth.ts';
|
|
import type { BaseBroadcaster } from '../services/broadcaster.ts';
|
|
import { ApplicationError, UnauthorizedError } from '../errors/index.ts';
|
|
import type { Database } from '../services/storage/database.ts';
|
|
import type { RouteDefinition, RouteModule, RouteStream } from './types.ts';
|
|
import type { Accounts } from '../auth/accounts.ts';
|
|
|
|
/**
|
|
* Schema to validate a single write resource.
|
|
*
|
|
* Each write targets one resource slot identified by (id, publicKey).
|
|
* The publicKey owner signs the payload to prove write authority.
|
|
*/
|
|
const writeResource = z.object({
|
|
id: z.string().min(1),
|
|
publicKey: z.string().min(1),
|
|
timestamp: z.number().positive(),
|
|
signature: z.string().min(1),
|
|
value: z.instanceof(Uint8Array),
|
|
});
|
|
|
|
/**
|
|
* Schema to validate the write body.
|
|
*
|
|
* The body contains an array of write resources.
|
|
*/
|
|
const writeBody = z.object({
|
|
resources: z.array(writeResource).min(1),
|
|
});
|
|
|
|
/**
|
|
* Schema to validate and deduplicate resource IDs supplied in an application body.
|
|
*
|
|
* The body contains an array of resource IDs.
|
|
*/
|
|
const resourceIdsSchema = z.object({
|
|
resourceId: z
|
|
.array(z.string().min(1))
|
|
.min(1, 'At least one resourceId is required')
|
|
.transform((ids) => [ ...new Set(ids) ]),
|
|
});
|
|
|
|
type WriteResource = z.infer<typeof writeResource>;
|
|
|
|
/**
|
|
* Resource data domain routes.
|
|
*
|
|
* A "resource" is a shared document identified by id. Multiple instances
|
|
* (one per publicKey) can coexist within the same resource. Clients read,
|
|
* write, subscribe to changes, and unsubscribe over the transport-neutral
|
|
* RouteStream API.
|
|
*/
|
|
export class DataRoute implements RouteModule {
|
|
constructor(
|
|
private readonly database: Database,
|
|
private readonly broadcaster: BaseBroadcaster,
|
|
private readonly auth: AuthSecp256k1,
|
|
private readonly accounts: Accounts,
|
|
) {}
|
|
|
|
/** Declare exact routes; each handler owns its stream behavior. */
|
|
async getRoutes(): Promise<Array<RouteDefinition>> {
|
|
return [
|
|
{
|
|
url: '/data/get',
|
|
handler: this.getData.bind(this),
|
|
},
|
|
{
|
|
url: '/data/write',
|
|
handler: this.writeData.bind(this),
|
|
},
|
|
{
|
|
url: '/data/subscribe',
|
|
handler: this.subscribeData.bind(this),
|
|
},
|
|
{
|
|
url: '/data/unsubscribe',
|
|
handler: this.unsubscribeData.bind(this),
|
|
},
|
|
];
|
|
}
|
|
|
|
/** Reads the requested resources from the database, returning all instances for each resource */
|
|
async getData(stream: RouteStream): Promise<void> {
|
|
const resourceIds = resourceIdsSchema.parse(stream.body).resourceId;
|
|
|
|
// Remove duplicates.
|
|
const uniqueIds = [ ...new Set(resourceIds) ];
|
|
|
|
// If there are no resource ids, return an empty array.
|
|
if (uniqueIds.length === 0) {
|
|
await stream.send([]);
|
|
|
|
return;
|
|
}
|
|
|
|
// Read the data from the database.
|
|
const rows = await this.database.db
|
|
.selectFrom('resource_data')
|
|
.select([ 'resource_id', 'public_key', 'blob', 'timestamp', 'signature' ])
|
|
.where('resource_id', 'in', uniqueIds)
|
|
.orderBy('timestamp', 'asc')
|
|
.execute();
|
|
|
|
// Format the rows into the read resource responses.
|
|
const formattedRows = rows.map((row) => ({
|
|
resourceId: row.resource_id,
|
|
publicKey: row.public_key,
|
|
blob: new Uint8Array(row.blob),
|
|
timestamp: row.timestamp,
|
|
signature: row.signature,
|
|
}));
|
|
|
|
await stream.send(formattedRows);
|
|
}
|
|
|
|
/**
|
|
* Authenticated batch write.
|
|
*
|
|
* Each resource in the batch is verified independently, then persisted and
|
|
* broadcast to any subscribers listening on that resource's topic.
|
|
*/
|
|
async writeData(stream: RouteStream): Promise<void> {
|
|
const { resources } = writeBody.parse(stream.body);
|
|
|
|
// temporarily disable the signature verification for testing
|
|
// const publicKey = stream.headers?.['x-public-key']
|
|
const publicKey = 'public-key';
|
|
|
|
// Authenticate the whole batch before producing any storage side effects.
|
|
// A single bad signature rejects the entire write — no partial commits.
|
|
await Promise.all(resources.map((resource) => this.verifyWriteResource(resource)));
|
|
|
|
const timestamp = Date.now();
|
|
const rows = resources.map((resource) => ({
|
|
resourceId: resource.id,
|
|
publicKey: resource.publicKey,
|
|
blob: Buffer.from(resource.value),
|
|
signature: resource.signature,
|
|
}));
|
|
|
|
// Get the total size of the bytes being written
|
|
const totalSize = rows.reduce((acc, row) => acc + row.blob.length, 0);
|
|
|
|
if (!await this.accounts.hasSufficientBalance(publicKey, totalSize)) {
|
|
throw new ApplicationError(203, 'Insufficient balance');
|
|
}
|
|
|
|
// Set the balance of the public key
|
|
await this.accounts.deductBalance(publicKey, totalSize);
|
|
|
|
// Upsert every row atomically; conflicts update blob and timestamp only.
|
|
await this.database.db.transaction().execute(async (trx) => {
|
|
for (const { resourceId, publicKey, blob, signature } of rows) {
|
|
await trx
|
|
.insertInto('resource_data')
|
|
.values({
|
|
resource_id: resourceId,
|
|
public_key: publicKey,
|
|
blob,
|
|
timestamp,
|
|
signature,
|
|
})
|
|
.onConflict((oc) =>
|
|
oc.columns([ 'resource_id', 'public_key' ]).doUpdateSet({
|
|
blob,
|
|
timestamp,
|
|
}))
|
|
.execute();
|
|
}
|
|
});
|
|
|
|
// Format the rows into the written resource responses.
|
|
const written = rows.map(({ resourceId, publicKey, blob, signature }) => ({
|
|
instance: {
|
|
resourceId,
|
|
publicKey,
|
|
blob: new Uint8Array(blob),
|
|
timestamp,
|
|
signature,
|
|
},
|
|
}));
|
|
|
|
// Notify subscribers on each changed resource. Topic names are scoped per
|
|
// resource id so clients only receive events for resources they joined.
|
|
for (const { instance } of written) {
|
|
await this.broadcaster.publish(DataRoute.resourceTopic(instance.resourceId), {
|
|
type: 'instance-changed',
|
|
data: instance,
|
|
});
|
|
}
|
|
|
|
// Return the persisted instances so the writer can confirm what was stored.
|
|
await stream.send({
|
|
resources: written.map(({ instance }) => ({
|
|
id: instance.resourceId,
|
|
...instance,
|
|
})),
|
|
balance: await this.accounts.getBalance(publicKey),
|
|
});
|
|
}
|
|
|
|
/** Subscribe this connection to future changes for the requested resources. */
|
|
async subscribeData(stream: RouteStream): Promise<void> {
|
|
const resourceIds = resourceIdsSchema.parse(stream.body).resourceId;
|
|
|
|
const topics = resourceIds.map((resourceId) => DataRoute.resourceTopic(resourceId));
|
|
|
|
// Subscribe registers the topics synchronously, then keeps this route
|
|
// active until those topics are removed or the connection closes.
|
|
await this.broadcaster.subscribe(stream, topics);
|
|
}
|
|
|
|
/**
|
|
* Leave resource topics without closing the connection.
|
|
*
|
|
* Only available on bidirectional transports (WebSocket). SSE clients
|
|
* unsubscribe implicitly by aborting the HTTP request.
|
|
*/
|
|
async unsubscribeData(stream: RouteStream): Promise<void> {
|
|
const resourceIds = resourceIdsSchema.parse(stream.body).resourceId;
|
|
|
|
// If the stream is not bidirectional, throw an error.
|
|
if (!stream.bidirectional) {
|
|
throw new ApplicationError(HTTP_STATUS_CODE_NOT_ACCEPTED, 'This route requires an existing bidirectional stream');
|
|
}
|
|
|
|
// Unsubscribe from the resource topics.
|
|
await this.broadcaster.unsubscribe(
|
|
stream,
|
|
resourceIds.map((resourceId) => DataRoute.resourceTopic(resourceId)),
|
|
);
|
|
await stream.send({});
|
|
}
|
|
|
|
/**
|
|
* Verify one write's timestamp freshness and secp256k1 signature.
|
|
*
|
|
* The client must sign the canonical payload built from timestamp, resource
|
|
* id, and value — not the raw HTTP/WebSocket envelope.
|
|
*/
|
|
private async verifyWriteResource(resource: WriteResource): Promise<void> {
|
|
this.auth.assertTimestampFreshness(resource.timestamp);
|
|
|
|
// Compile the signature payload as `Timestamp:ID:Value`
|
|
const signaturePayload = `${resource.timestamp}:${resource.id}:${toExtendedJson(resource.value)}`;
|
|
|
|
if (!(await this.auth.verifySignature(resource.publicKey, resource.signature, signaturePayload))) {
|
|
throw new UnauthorizedError('Invalid resource signature');
|
|
}
|
|
}
|
|
|
|
/** Broadcaster topic for a single resource's instance-changed events. */
|
|
private static resourceTopic(resourceId: string): string {
|
|
return `resource:${resourceId}`;
|
|
}
|
|
}
|