nostr_event_scheduler

Local-first Dart package for scheduling Nostr events via Scheduler DVMs.

This package implements the Scheduler DVM protocol and provides a robust, offline-first API for creating, tracking, and cancelling scheduled Nostr events.

Features

• Local-first - Every operation is persisted locally before any network attempt
• Multi-account - Every call takes an explicit account pubkey, so one instance serves as many accounts as you load in ndk.accounts
• Offline signer support - Works even when your signer (e.g. NIP-46) is temporarily unavailable
• Multi-device sync - Automatically syncs scheduled jobs across devices
• Redundant scheduling - Send the same job to several Scheduler DVMs (one kind:5905 per DVM, same job_id) so publishing does not depend on a single DVM's uptime or policy
• Scheduled packages - Group several DVM jobs into one logical schedule with private display context
• Real-time DVM feedback - Receives status updates from Scheduler DVMs (scheduled, published, failed, etc.)
• No raw event duplication - Relies on the NDK persistent cache for raw events; only stores decrypted payloads, tombstones and projections in Sembast
• No migrations - A schema change drops the projections and recomputes them from raw, offline
• Controlled network access - Explicit per-account startListening / stopListening for fine-grained relay connectivity control

Quick start

import 'package:broadcast_queue_shim_for_ndk/broadcast_queue_shim_for_ndk.dart';
import 'package:ndk/ndk.dart';
import 'package:nostr_event_scheduler/nostr_event_scheduler.dart';
import 'package:sembast/sembast_io.dart';

Future<void> main() async {
  final db = await databaseFactoryIo.openDatabase('scheduler.db');

  // The cache must be persistent: it holds the raw events the scheduler
  // recomputes its projections from.
  final ndk = Ndk(
    NdkConfig(
      eventVerifier: Bip340EventVerifier(),
      cache: SembastCacheManager(db),
      fetchedRangesEnabled: true,
    ),
  );

  final broadcast = OfflineBroadcast.withNdk(ndk, db: db);
  broadcast.start();

  final scheduler = EventScheduler(
    ndk: ndk,
    broadcast: broadcast,
    db: db,
  );

  await scheduler.startListening(pubkey: myPubKey);

  // Listen to status updates from the DVM
  scheduler.statusUpdates.listen((update) {
    print('Job ${update.jobId} of ${update.pubkey}: ${update.status}');
  });

  // Schedule an event
  final scheduleAt = DateTime.now().add(const Duration(hours: 1));

  final event = Nip01Event(
    pubKey: myPubKey,
    kind: 1,
    tags: [],
    content: 'Hello from the future!',
    createdAt: scheduleAt.millisecondsSinceEpoch ~/ 1000,
  );
  final signedEvent = await ndk.accounts.getLoggedAccount()!.signer.sign(event);

  final job = await scheduler.schedule(
    signedEvent,
    [dvmPubkey],
    pubkey: myPubKey,
    at: scheduleAt,
    relays: ['wss://relay.damus.io'],
  );

  print('Scheduled job: ${job.jobId}');

  // List several DVMs to schedule redundantly: one kind:5905 per DVM, all
  // sharing the same job_id and payload. One publication is enough, relays
  // deduplicate the signed event by ID.
  final signedEventR = await ndk.accounts.getLoggedAccount()!.signer.sign(
    Nip01Event(
      pubKey: myPubKey,
      kind: 1,
      tags: [],
      content: 'Published even if one DVM is down',
      createdAt: scheduleAt.millisecondsSinceEpoch ~/ 1000,
    ),
  );

  final redundantJob = await scheduler.schedule(
    signedEventR,
    [dvmPubkey, anotherDvmPubkey],
    pubkey: myPubKey,
    at: scheduleAt,
    relays: ['wss://relay.damus.io'],
  );

  // One logical job, one request per DVM, aggregated status
  print('Job ${redundantJob.jobId} via ${redundantJob.dvmPubkeys.length} DVMs');
  print('Status: ${redundantJob.status}');

  // List all jobs
  final jobs = await scheduler.listJobs(pubkey: myPubKey);
  print('Total jobs: ${jobs.length}');

  // Group multiple DVM jobs as one logical schedule
  final signedEventB = await ndk.accounts.getLoggedAccount()!.signer.sign(
    Nip01Event(
      pubKey: myPubKey,
      kind: 1,
      tags: [],
      content: 'Package item B',
      createdAt: scheduleAt.millisecondsSinceEpoch ~/ 1000,
    ),
  );
  final signedEventC = await ndk.accounts.getLoggedAccount()!.signer.sign(
    Nip01Event(
      pubKey: myPubKey,
      kind: 1,
      tags: [],
      content: 'Package item C',
      createdAt:
          scheduleAt.add(const Duration(minutes: 5)).millisecondsSinceEpoch ~/
          1000,
    ),
  );

  final package = await scheduler.schedulePackage(
    [
      SchedulePackageItem(
        event: signedEventB,
        // A package item can also fan out to several DVMs
        dvmPubkeys: [dvmPubkey, anotherDvmPubkey],
        at: scheduleAt,
        relays: ['wss://relay.damus.io'],
      ),
      SchedulePackageItem(
        event: signedEventC,
        dvmPubkeys: [anotherDvmPubkey],
        at: scheduleAt.add(const Duration(minutes: 5)),
        relays: ['wss://nos.lol'],
        dvmReadRelays: ['wss://dvm-inbox.example'],
      ),
    ],
    content: 'Private app context for displaying this package later',
    pubkey: myPubKey,
  );

  print('Scheduled package: ${package.packageId}');

  // List logical schedules: standalone jobs + packages
  final schedules = await scheduler.listSchedules(pubkey: myPubKey);
  print('Total schedules: ${schedules.length}');

  // Cancel a job
  await scheduler.cancel(job.jobId, pubkey: myPubKey);

  // Cancel a package and all linked DVM jobs
  await scheduler.cancelPackage(package.packageId, pubkey: myPubKey);

  // Wipe everything this account stored locally
  await scheduler.clearLocalAccountData(pubkey: myPubKey);

  // Dispose when done
  await scheduler.dispose();
  await broadcast.dispose();
  await db.close();
}

API Overview

EventScheduler

The main entry point.

Every method takes the account it acts for. The signer is resolved from ndk.accounts, so the logged account is never used implicitly.

clearLocalAccountData is a local reset, not a protocol-level forget: the requests still live on the relays, so an account whose signer is still loaded rebuilds them on its next resync(). Use cancel to actually retract a schedule.

Models

• ScheduledJob - One logical scheduled event, with one request per DVM and an aggregated status
• ScheduledJobRequest - One kind:5905 request to a single DVM, with the status that DVM reported
• SchedulePackageItem - Input model for one job inside schedulePackage
• ScheduledPackage - Represents a package manifest and its linked jobs
• ScheduledItem - Logical schedule item, either a standalone job or a package
• JobStatus - Enum: pending, scheduled, published, failed, cancelled, error; JobStatus.aggregate combines the statuses of one job's requests
• StatusUpdate - Emitted when a DVM feedback is received, with the reporting DVM's pubkey and the owning account
• SyncState - Tracks whether one account's local state is up-to-date with the network

Architecture

The package follows a strict raw vs computed architecture:

• Raw holds definitive facts. The signed events (kind:5905, kind:31234, kind:5, kind:7000) live in the NDK persistent cache; their decrypted payloads and deletion tombstones live in Sembast, keyed by event id. Raw is never dropped and never migrated.
• Computed holds the projections (jobs, packages, and the pending decryption queue), in Sembast, each record tagged with its owning account.

A schema change therefore needs no migration script: the projections are dropped and recomputed from raw, per account and lazily, without network access or user action. This does mean the host must give Ndk a persistent CacheManager. See ARCHITECTURE.md for the full design document.

Testing

The package includes integration tests using a minimal MockRelay implementation and an in-process nostr_scheduler_dvm instance.

dart test