@netscript/plugin-streams

Durable Streams protocol plugin for NetScript: a plugin manifest plus CLI, scaffolding, end-to-end gate, and Aspire integration surfaces for an ephemeral change-data stream service with startup-probed file storage as an explicit opt-in supporting orderly process restart. This page is written against the plugin public surface reported by deno doc. For the full index of packages and plugins return to the reference overview.

The plugin ships seven published entrypoints. The root export (@netscript/plugin-streams) carries the manifest and the typed topic definition helper; six sub-path exports carry the framework integrations:

• @netscript/plugin-streams/cli — plugin CLI command group.
• @netscript/plugin-streams/adapter-cli — executable plugin-adapter CLI entrypoint.
• @netscript/plugin-streams/scaffold — scaffolder protocol entrypoint.
• @netscript/plugin-streams/e2e — E2E gate definitions.
• @netscript/plugin-streams/aspire — Aspire AppHost contribution.
• @netscript/plugin-streams/services — executable Durable Streams service.

The schema, producer, telemetry, testing, and diagnostics primitives that the plugin builds on live in the separately published @netscript/plugin-streams-core package, which has its own canonical reference page.

Plugin manifest

Topic authoring

Runtime producers live in @netscript/plugin-streams-core: use defineStreamSchema with createDurableStream or createServiceStreamProducer. The former root producer/consumer helpers and their unsupported-operation types were removed in 0.0.8. See the breaking migration for the collection model and HTTP/SSE consumption path.

Browser consumers read over HTTP/SSE. There is no in-process subscribe(); a browser consumes a stream by reading its HTTP/SSE endpoint (for example with EventSource). The local AppHost serves the generated app over http://, and under HTTP/1.1 browsers typically allow only about six concurrent connections per origin — so several long-lived stream subscriptions from one page can starve later requests. Serve over HTTPS (HTTP/2) when a page needs many simultaneous stream consumers.

Re-exported plugin framework types

The root export re-exports the shared NetScript plugin-framework contribution types unchanged from @netscript/plugin. See the @netscript/plugin reference for their full definitions.

Sub-path exports

@netscript/plugin-streams/cli

CLI command group for the plugin, mounted under deno x -A jsr:@netscript/plugin-streams@0.0.7/cli.

The CLI types PluginCli, PluginCliArgs, PluginCliCommand, and PluginCliResult are re-exported from @netscript/plugin and document the base command contract StreamsCli extends.

@netscript/plugin-streams/scaffold

@netscript/plugin-streams/e2e

@netscript/plugin-streams/aspire

The Aspire entrypoint contributes the ephemeral-by-default Durable Streams protocol service to an Aspire AppHost. File storage requires an explicit, existing writable directory and a successful startup write/read probe; invalid opt-ins fail startup.

The remaining symbols on this entrypoint (AspireBuilder, AspireResource, AspireResourceKind, AspireNSPluginContribution, ContributionContext, CacheSpec, ContainerSpec, DatabaseSpec, DenoServiceSpec, DenoBackgroundSpec, EnvSource) are re-exported unchanged from @netscript/aspire; see that reference page for their definitions.

Core package

The separately published @netscript/plugin-streams-core page is canonical for stream schemas, producers, configuration, telemetry, testing, and diagnostics. This page stays focused on the deployable plugin's manifest and integration entrypoints. The testing example below uses core APIs intentionally; exhaustive core entrypoint and symbol tables live only on its reference page.

Socket-Free Testing and Telemetry

Use MemoryStreamProducer in unit and integration tests to verify change events are published without opening network sockets or starting a real streams server. Use createStreamsInstrumentation (from ./telemetry) to verify context-propagation headers (like OpenTelemetry traceparent) are correctly injected.

import { assertEquals, assertExists } from "@std/assert";
import { MemoryStreamProducer } from "@netscript/plugin-streams-core/testing";
import { createStreamsInstrumentation } from "@netscript/plugin-streams-core/telemetry";

Deno.test("stream socket-free testing with MemoryStreamProducer and telemetry facade", async () => {
  // 1. Socket-free testing using MemoryStreamProducer
  const producer = new MemoryStreamProducer();

  // Publish some mock events
  producer.upsert("orders", { id: "order_123", total: 100 });
  producer.delete("orders", "order_123");
  await producer.flush();

  // Assert events are recorded synchronously in memory without network sockets
  assertEquals(producer.events(), [
    {
      entityType: "orders",
      operation: "upsert",
      key: "order_123",
      value: { id: "order_123", total: 100 },
    },
    {
      entityType: "orders",
      operation: "delete",
      key: "order_123",
    },
  ]);

  await producer.close();

  // 2. Using the Telemetry Facade to publish and inject trace context
  const telemetry = createStreamsInstrumentation();
  let traceHeaders: Record<string, string> = {};

  telemetry.publish({
    streamPath: "/orders/updates",
    collection: "orders",
    operation: "upsert",
    producerId: "orders-service",
    messageId: "order_123",
    emit: (headers) => {
      traceHeaders = headers;
    },
  });

  // Assert traceparent is generated and injected by the telemetry facade
  assertExists(traceHeaders.traceparent);
});

Back to the reference overview.