feat: claim offline receives from encrypted funding packets

.github/workflows/trivy.yaml

@@ -29,8 +29,6 @@ jobs:
 
       - name: Run Trivy vulnerability scanner
         uses: aquasecurity/trivy-action@57a97c7e7821a5776cebc9bb87c984fa69cba8f1 # 0.35.0
-        env:
-          TRIVY_DB_REPOSITORY: public.ecr.aws/aquasecurity/trivy-db:2
         with:
           image-ref: "${{ env.IMAGE_NAME }}:${{ github.sha }}"
           format: "table"

README.md

@@ -7,7 +7,7 @@ owner is offline. A wallet registers a watch on its address; every spend is
 co-signed by the [emulator](https://github.com/arkade-os/emulator) only if it
 satisfies the template's covenant.
 
-The default templates in [`templates/`](templates) (renewal, boarding, VHTLC claim and refund, and noninteractiveclaim) are registered and trusted at startup.
+The default templates in [`templates/`](templates) (renewal, boarding, VHTLC claim and refund, noninteractiveclaim, and offline_receive) are registered and trusted at startup.
 
 ## Run
 
@@ -79,5 +79,73 @@ belongs to one pool, and each pool has one output without an `amount`.
 
 ## Stream delegations
 
-A template with a `trigger` is not registered by clients: the operator registers it as a stream delegation (admin `RegisterStreamDelegation`; a default with a trigger starts by itself when an encryption key is configured). The daemon reads arkd's transaction stream and, for every transaction whose extension holds the trigger's packet, binds the trigger's variables, checks the template locks the output, and spawns a watch of the template that expires after `expires_after_seconds`.
-
+Offline receive uses `templates/offline_receive.json` as an operator stream delegation.
+The receiver never registers a watch or sends its preimage to a delegatee API. Before
+requesting an exact-output quote, it seals the preimage, recipient and minimum amount
+to the daemon's advertised encryption public key. The operator can decrypt the preimage
+and must be trusted with it. The funding transaction delivers
+that ciphertext in extension packet type 4. The solver adds its payout script in TLV 2;
+the daemon checks the full funding taptree and output script against the resulting
+covenant claim leaf, and authenticates its recipient against the envelope before claiming.
+
+The packet body starts with TLV 127 containing ASCII `delegatee-secrets/v1`, followed
+by TLV 1 (ciphertext) and TLV 3 (compressed encryption public key). Each record is a
+one-byte tag, two-byte big-endian length and value. The ciphertext uses the existing
+ECIES wire format and `covclaimd/preimage/v1` domain, encrypting JSON:
+`{"format":"delegatee-secrets/v1","template_id":"<content ID>","values":{...}}`.
+The values are `preimage` and `receiver_program` (32-byte lowercase hex) and
+`minimum_value` (canonical ScriptNum hex). Secret declarations select fields with
+`field`. The recipient captured from TLV 2 must equal the authenticated receiver before
+a claim can disclose the preimage. Raw ECIES plaintext remains supported by
+templates without `field`, including the unchanged `noninteractiveclaim` stream.
+
+The authenticated minimum must be positive, at most 2.1e15 satoshis, and satisfied by
+the funded input before a claim can disclose the preimage. The output preserves the
+input's entire value and assets and pays the authenticated recipient. This envelope
+binds an intent before its quote, not a specific later refund deadline or lockup tree.
+There is no encrypted claim deadline; ordinary spawned watches expire after 24 hours.
+
+The receiver can go offline after preparing its invoice. Once the daemon receives the
+funding packet, the spawned watch persists and resumes retries after a daemon restart.
+The filtered indexer subscription does not replay history: funding first seen while the daemon
+is unavailable cannot currently be discovered through this packet-only flow. The
+indexer recovers known watches, not entirely unseen funding packets.
+
+A template with a `trigger` is not registered by clients: the operator registers it as a stream delegation (admin `RegisterStreamDelegation`; a default with a trigger starts by itself when an encryption key is configured). The daemon opens an indexer subscription with CEL filters derived from the active triggers. For offline receive these select packet type 4, the version marker and one of its retained encryption keys. It binds matching packets, checks the template locks the output, and spawns a watch that expires after `expires_after_seconds`. Full local trigger and covenant checks remain authoritative: the CEL filter only reduces traffic. This requires indexer initial-filter support; subscription errors are reported and retried without switching to the global transaction stream.
+
+
+## Capacity
+
+Discovery filter changes update the existing subscription atomically while its
+events continue to be processed. Failed or ambiguous updates are reconciled
+without closing the working stream.
+
+Known offchain direct watches use script subscriptions, split into at most 1,000
+scripts per subscription. Existing shards retain their scripts as registrations
+change. New and spent coin events invalidate only the affected scripts; repeated
+events coalesce. The next scan queries those scripts in chunks of 100 with at
+most four requests in flight. Startup, connection gaps and periodic reconciliation
+refresh all active scripts. Frequent events cannot postpone the periodic refresh.
+Upstream throttle backoff takes precedence over event wakes and reconciliation
+deadlines; failed scans retain their dirty scripts for the next permitted retry.
+
+Packet discovery needs a recipient filter because a new payment's script is not
+known beforehand. It uses one separate filtered subscription, with a one-event
+application buffer and synchronous registration. This bounds memory, but does not
+provide durable event replay or guaranteed admission during overload, a database
+outage or a full registration cap. Watch rows already persisted are recoverable.
+
+The default caps are 50,000 public/stream registrations and 10,000 spawned watches.
+A successful receive watch remains active until its 24-hour expiry so it can handle
+another funding to that script. Consequently 10,000 distinct receives can fill the
+child cap during a day even when claims settle quickly. Raising the cap requires
+capacity testing; it does not increase processing throughput.
+
+All active registrations are still loaded and examined each scan, and direct claims
+run serially within their lane. Registration inserts serialize and count active
+rows. Batch planning uses sorted passes rather than repeatedly copying/scanning
+remaining entries. Unit scale tests cover 50,000 scripts, bounded subscriptions,
+coalesced refreshes, races and failed/reconnected streams; these use mock storage
+and indexer responses and do not establish production throughput. Measure database
+latency, scan duration, claim service time, oldest pending receive and cap usage
+under the deployment's actual traffic before increasing limits.

go.mod

@@ -7,6 +7,7 @@ replace github.com/arkade-os/delegatee/api-spec => ./api-spec
 replace github.com/arkade-os/delegatee/pkg/template => ./pkg/template
 
 require (
+	github.com/arkade-os/arkd/api-spec v0.0.0-20260901090427-f863e4847193
 	github.com/arkade-os/arkd/pkg/ark-lib v0.8.1-0.20260901090427-f863e4847193
 	github.com/arkade-os/arkd/pkg/client-lib v0.0.0-20260901090427-f863e4847193
 	github.com/arkade-os/arkd/pkg/client-wallet v0.0.0-20260901090427-f863e4847193
@@ -40,7 +41,6 @@ require (
 	cel.dev/expr v0.25.2 // indirect
 	github.com/aead/siphash v1.0.1 // indirect
 	github.com/antlr4-go/antlr/v4 v4.13.1 // indirect
-	github.com/arkade-os/arkd/api-spec v0.0.0-20260901090427-f863e4847193 // indirect
 	github.com/arkade-os/arkd/pkg/errors v0.0.0-20260901090427-f863e4847193 // indirect
 	github.com/arkade-os/emulator/api-spec v0.1.0 // indirect
 	github.com/beorn7/perks v1.0.1 // indirect

internal/config/config.go

@@ -14,11 +14,11 @@ import (
 
 	clientlib "github.com/arkade-os/arkd/pkg/client-lib"
 	"github.com/arkade-os/arkd/pkg/client-lib/client"
-	"github.com/arkade-os/arkd/pkg/client-lib/indexer"
 	"github.com/arkade-os/delegatee/internal/core/application"
 	"github.com/arkade-os/delegatee/internal/core/domain"
 	"github.com/arkade-os/delegatee/internal/infrastructure/db/postgres"
 	"github.com/arkade-os/delegatee/internal/infrastructure/explorer"
+	"github.com/arkade-os/delegatee/internal/infrastructure/indexer"
 	"github.com/arkade-os/delegatee/templates"
 	emulatorclient "github.com/arkade-os/emulator/pkg/client"
 	"github.com/btcsuite/btcd/btcec/v2"

internal/config/config_test.go

@@ -255,12 +255,13 @@ func TestRegisterDefaults(t *testing.T) {
 	renewal, boarding := "d901cb8554a77524fb55ebfcde8e1c468871de45163abf34e988c47f1fa36896", "02b4e8bdfb52d7450ab75510cfaff36da5b88323291325e38358df10684a8344"
 	claim, refund := "c6e10a94e6b27b1c29dc3c9eae95d1f01ef9e4142f882f9ce9888a416d7ae01d", "978df861b2ae33f1373fd9ead844b074b3575d2341e98f1d50114179a9ed7d6d"
 	noninteractiveclaim := "4493579d4cdb14a5809b55c55b7d228a4d8f7fc479c1424b286d4d5234382eee"
+	offlineReceive := "fe39de17a0f608b710d9290b7533ec48204493a3e7cf726e03098b996eea38a2"
 	on := &Config{DefaultTemplates: true}
 	svc := newDefaultsService()
 	require.NoError(t, on.registerDefaults(t.Context(), svc))
 	require.Len(t, svc.artifacts, 1)
-	require.Equal(t, map[string]bool{renewal: true, boarding: true, claim: true, refund: true, noninteractiveclaim: true}, svc.trusted())
-	require.Equal(t, map[string]bool{noninteractiveclaim: true}, svc.streams, "only the template with a trigger streams")
+	require.Equal(t, map[string]bool{renewal: true, boarding: true, claim: true, refund: true, noninteractiveclaim: true, offlineReceive: true}, svc.trusted())
+	require.Equal(t, map[string]bool{noninteractiveclaim: true, offlineReceive: true}, svc.streams, "templates with triggers stream")
 
 	noKey := newDefaultsService()
 	noKey.streamErr = application.ErrSecretsRequired
@@ -270,7 +271,7 @@ func TestRegisterDefaults(t *testing.T) {
 	trusts := svc.trusts
 	require.NoError(t, on.registerDefaults(t.Context(), svc))
 	require.Equal(t, trusts, svc.trusts, "a restart changes nothing")
-	require.Equal(t, map[string]bool{renewal: true, boarding: false, claim: true, refund: true, noninteractiveclaim: true}, svc.trusted(), "an untrusted default stays untrusted")
+	require.Equal(t, map[string]bool{renewal: true, boarding: false, claim: true, refund: true, noninteractiveclaim: true, offlineReceive: true}, svc.trusted(), "an untrusted default stays untrusted")
 
 	off := newDefaultsService()
 	require.NoError(t, (&Config{}).registerDefaults(t.Context(), off))

internal/core/application/collection.go

@@ -60,19 +60,29 @@ func (s *service) planBatches(entries []planEntry, now time.Time) *plan {
 	slices.SortStableFunc(vtxos, func(a, b planEntry) int {
 		return cmp.Or(a.deadline.Compare(b.deadline), a.due.Compare(b.due))
 	})
-	// deadline >= due: the entry with the earliest deadline is always in its batch, so this ends
-	for len(vtxos) > 0 {
-		at := vtxos[0].deadline
-		b := PlannedBatch{At: at, ForcedBy: vtxos[0].outpoints[0], ForcedByDelegation: vtxos[0].delegation}
-		var rest []planEntry
-		for _, e := range vtxos {
+	dueOrder := make([]int, len(vtxos))
+	for i := range dueOrder {
+		dueOrder[i] = i
+	}
+	slices.SortFunc(dueOrder, func(a, b int) int { return vtxos[a].due.Compare(vtxos[b].due) })
+	taken := make([]bool, len(vtxos))
+	nextDue := 0
+	for i, first := range vtxos {
+		if taken[i] {
+			continue
+		}
+		at := first.deadline
+		b := PlannedBatch{At: at, ForcedBy: first.outpoints[0], ForcedByDelegation: first.delegation}
+		for nextDue < len(dueOrder) {
+			index := dueOrder[nextDue]
+			e := vtxos[index]
 			if e.due.After(at) {
-				rest = append(rest, e)
-				continue
+				break
 			}
 			p.join(&b, e, at)
+			taken[index] = true
+			nextDue++
 		}
-		vtxos = rest
 		p.batches = append(p.batches, b)
 	}
 	for _, e := range deposits {

internal/core/application/collection_test.go

@@ -1,7 +1,11 @@
 package application
 
 import (
+	"cmp"
 	"errors"
+	"fmt"
+	"math/rand/v2"
+	"slices"
 	"testing"
 	"testing/synctest"
 	"time"
@@ -234,3 +238,54 @@ func TestReserveMustCoverARetry(t *testing.T) {
 		[]*btcec.PrivateKey{key}, time.Hour, time.Minute, 2*time.Minute, 0, env.svc.limits)
 	require.NoError(t, err)
 }
+
+func BenchmarkPlanScatteredDeadlines(b *testing.B) {
+	now := time.Now()
+	entries := make([]planEntry, 5000)
+	for n := range entries {
+		at := now.Add(time.Duration(n+1) * time.Second)
+		entries[n] = planEntry{outpoints: []string{fmt.Sprintf("coin-%d", n)}, delegation: int64(n), amount: 1000, due: at, deadline: at}
+	}
+	svc := &service{}
+	b.ReportAllocs()
+	b.ResetTimer()
+	for range b.N {
+		plan := svc.planBatches(entries, now)
+		if len(plan.batches) != len(entries) {
+			b.Fatal("scattered deadlines lost")
+		}
+	}
+}
+
+func TestPlanBatchSweepPreservesAssignments(t *testing.T) {
+	rng := rand.New(rand.NewPCG(1, 2))
+	now := time.Now()
+	svc := &service{boardingMaxWait: 10 * time.Minute}
+	for range 100 {
+		entries := make([]planEntry, 100)
+		for i := range entries {
+			due := now.Add(time.Duration(rng.IntN(100)-20) * time.Minute)
+			entries[i] = planEntry{outpoints: []string{fmt.Sprintf("coin-%d", i)}, delegation: int64(i), amount: uint64(i + 1), due: due, deadline: due.Add(time.Duration(rng.IntN(20)) * time.Minute)}
+		}
+		remaining := slices.Clone(entries)
+		slices.SortStableFunc(remaining, func(a, b planEntry) int { return cmp.Or(a.deadline.Compare(b.deadline), a.due.Compare(b.due)) })
+		expected := &plan{batchAt: map[string]time.Time{}}
+		for len(remaining) > 0 {
+			first := remaining[0]
+			at := first.deadline
+			batch := PlannedBatch{At: at, ForcedBy: first.outpoints[0], ForcedByDelegation: first.delegation}
+			var later []planEntry
+			for _, e := range remaining {
+				if e.due.After(at) {
+					later = append(later, e)
+				} else {
+					expected.join(&batch, e, at)
+				}
+			}
+			batch.At = maxTime(batch.At, now)
+			expected.batches = append(expected.batches, batch)
+			remaining = later
+		}
+		require.Equal(t, expected, svc.planBatches(entries, now))
+	}
+}

internal/core/application/discovery.go

@@ -276,7 +276,7 @@ func (s *service) discover(ctx context.Context, delegations []domain.Delegation,
 	unwatched := map[int64]string{}
 	s.found = map[int64][]scannedCoin{}
 	scripts := s.watchAll(ctx, delegations, active, unwatched)
-	byScript, err := s.spendableVtxos(ctx, scripts)
+	byScript, err := s.scanVtxos(ctx, scripts)
 	if err != nil {
 		return nil, nil, nil, err
 	}