vtxo: event and persistence projection for cosigned on-chain spends (M3 slice e)

internal/core/application/indexer_test.go

@@ -1232,6 +1232,15 @@ func (m *mockVtxoRepoForIndexer) SpendVtxos(
 	return nil
 }
 
+func (m *mockVtxoRepoForIndexer) RecordCosignedTx(
+	ctx context.Context,
+	txid string,
+	inputs []domain.Outpoint,
+	outputs []domain.Vtxo,
+) error {
+	return nil
+}
+
 func (m *mockVtxoRepoForIndexer) UnrollVtxos(
 	ctx context.Context,
 	outpoints []domain.Outpoint,

internal/core/application/sweeper_test.go

@@ -420,6 +420,12 @@ func (m *mockVtxoRepository) SpendVtxos(
 ) error {
 	return nil
 }
+func (m *mockVtxoRepository) RecordCosignedTx(
+	ctx context.Context, txid string, inputs []domain.Outpoint, outputs []domain.Vtxo,
+) error {
+	return nil
+}
+
 func (m *mockVtxoRepository) UnrollVtxos(ctx context.Context, outpoints []domain.Outpoint) error {
 	return nil
 }

internal/core/domain/vtxo.go

@@ -69,16 +69,30 @@ type Vtxo struct {
 }
 
 // VtxoKind distinguishes how a vtxo is held. Offchain (the default) is a batch
-// leaf or an offchain-tx output. Onchain marks a vtxo held in an on-chain
-// Arkade UTXO. It is an open enum so future on-chain sub-kinds can be added
+// leaf or an offchain-tx output. The on-chain kinds mark a vtxo held in an
+// on-chain Arkade UTXO. It is an open enum so further sub-kinds can be added
 // without another schema migration.
 type VtxoKind uint8
 
 const (
 	VtxoKindOffchain VtxoKind = iota
 	VtxoKindOnchain
+	// VtxoKindOnchainPending is an output of a transaction arkd cosigned that
+	// has not confirmed. It exists so the indexer can show the output during
+	// the confirmation window without claiming it is spendable. The
+	// per-transaction lifecycle promotes it to VtxoKindOnchain on confirmation
+	// and deletes it if the transaction dies.
+	VtxoKindOnchainPending
 )
 
+// IsOnchainKind reports a vtxo held in an on-chain Arkade UTXO, confirmed or
+// still pending. Every rule that turns on the distinction between on-chain and
+// off-chain has to ask this rather than compare against a single kind, since a
+// pending output is as far from a batch leaf as a confirmed one is.
+func (v Vtxo) IsOnchainKind() bool {
+	return v.Kind == VtxoKindOnchain || v.Kind == VtxoKindOnchainPending
+}
+
 func (v Vtxo) String() string {
 	// nolint
 	b, _ := json.MarshalIndent(v, "", "  ")
@@ -88,14 +102,14 @@ func (v Vtxo) String() string {
 func (v Vtxo) IsNote() bool {
 	// An on-chain Arkade UTXO also has no commitment txids, so the kind check
 	// keeps it from reading as a note.
-	return v.Kind != VtxoKindOnchain &&
+	return !v.IsOnchainKind() &&
 		len(v.CommitmentTxids) <= 0 && v.RootCommitmentTxid == ""
 }
 
 func (v Vtxo) RequiresForfeit() bool {
 	// An on-chain Arkade UTXO joins a batch as a boarding input, which is
 	// signed directly and never forfeited.
-	return v.Kind != VtxoKindOnchain && !v.Swept && !v.IsNote() && !v.Unrolled
+	return !v.IsOnchainKind() && !v.Swept && !v.IsNote() && !v.Unrolled
 }
 
 func (v Vtxo) IsSettled() bool {
@@ -123,15 +137,15 @@ func (v Vtxo) OutputScript() ([]byte, error) {
 // comparison against a real deadline. Notes read as having one. They have no
 // batch either, but never reach a raw reader of the field.
 func (v Vtxo) HasBatchExpiry() bool {
-	return v.Kind != VtxoKindOnchain
+	return !v.IsOnchainKind()
 }
 
 func (v Vtxo) IsExpired() bool {
 	// An on-chain Arkade UTXO has no batch expiry, so ExpiresAt is not
 	// meaningful for it. Without this an on-chain vtxo (which carries a zero
 	// ExpiresAt) would read as permanently expired and be treated as
 	// unspendable by every caller.
-	if v.Kind == VtxoKindOnchain {
+	if v.IsOnchainKind() {
 		return false
 	}
 	return time.Now().After(time.Unix(v.ExpiresAt, 0))

internal/core/domain/vtxo_repo.go

@@ -7,6 +7,16 @@ type VtxoRepository interface {
 	SettleVtxos(ctx context.Context, spentVtxos map[Outpoint]string, commitmentTxid string) error
 	SpendVtxos(ctx context.Context, spentVtxos map[Outpoint]string, arkTxid string) error
 	UnrollVtxos(ctx context.Context, outpoints []Outpoint) error
+	// RecordCosignedTx records a transaction arkd cosigned in one durable write.
+	// The inputs are marked spent by txid and left with no ark txid, which is
+	// what distinguishes an onchain spend from an in-Ark one. The outputs are
+	// inserted as VtxoKindOnchainPending regardless of the kind the caller set,
+	// since nothing arkd has only cosigned is spendable yet.
+	//
+	// Both halves land together or not at all. The caller holds a claim on the
+	// inputs across this write and releases it afterwards, so a partial write
+	// would free an input whose spend was never recorded.
+	RecordCosignedTx(ctx context.Context, txid string, inputs []Outpoint, outputs []Vtxo) error
 	GetVtxos(ctx context.Context, outpoints []Outpoint) ([]Vtxo, error)
 	GetAllNonUnrolledVtxos(ctx context.Context, pubkey string) ([]Vtxo, []Vtxo, error)
 	GetAllSweepableUnrolledVtxos(ctx context.Context) ([]Vtxo, error)

internal/core/domain/vtxo_test.go

@@ -75,6 +75,11 @@ func TestVtxo_IsNote(t *testing.T) {
 			},
 			isNote: false,
 		},
+		{
+			name:   "pending onchain kind is not a note either",
+			vtxo:   domain.Vtxo{Kind: domain.VtxoKindOnchainPending},
+			isNote: false,
+		},
 	}
 	for _, f := range fixtures {
 		t.Run(f.name, func(t *testing.T) {
@@ -83,6 +88,38 @@ func TestVtxo_IsNote(t *testing.T) {
 	}
 }
 
+// Every rule keyed on the kind has to treat a pending output the same as a
+// confirmed one. A comparison against VtxoKindOnchain alone would let a pending
+// output read as a note, require a forfeit, and expire at the epoch.
+func TestVtxo_IsOnchainKind(t *testing.T) {
+	fixtures := []struct {
+		name string
+		vtxo domain.Vtxo
+		want bool
+	}{
+		{
+			name: "a batch vtxo is not onchain",
+			vtxo: domain.Vtxo{},
+			want: false,
+		},
+		{
+			name: "a confirmed onchain vtxo is",
+			vtxo: domain.Vtxo{Kind: domain.VtxoKindOnchain},
+			want: true,
+		},
+		{
+			name: "a pending onchain vtxo is too",
+			vtxo: domain.Vtxo{Kind: domain.VtxoKindOnchainPending},
+			want: true,
+		},
+	}
+	for _, f := range fixtures {
+		t.Run(f.name, func(t *testing.T) {
+			require.Equal(t, f.want, f.vtxo.IsOnchainKind())
+		})
+	}
+}
+
 func TestVtxo_IsSettled(t *testing.T) {
 	fixtures := []struct {
 		name      string
@@ -123,6 +160,11 @@ func TestVtxo_HasBatchExpiry(t *testing.T) {
 			vtxo: domain.Vtxo{Kind: domain.VtxoKindOnchain},
 			want: false,
 		},
+		{
+			name: "a pending onchain-kind vtxo has none either",
+			vtxo: domain.Vtxo{Kind: domain.VtxoKindOnchainPending},
+			want: false,
+		},
 		{
 			// The kind decides, not the value.
 			name: "a zero expiry on a batch vtxo still counts",
@@ -163,6 +205,11 @@ func TestVtxo_IsExpired(t *testing.T) {
 			},
 			isExpired: false,
 		},
+		{
+			name:      "pending onchain kind never expires either",
+			vtxo:      domain.Vtxo{Kind: domain.VtxoKindOnchainPending, ExpiresAt: 1},
+			isExpired: false,
+		},
 	}
 	for _, f := range fixtures {
 		t.Run(f.name, func(t *testing.T) {
@@ -233,6 +280,11 @@ func TestVtxo_RequiresForfeit(t *testing.T) {
 			},
 			requiresForfeit: false,
 		},
+		{
+			name:            "should be false (pending onchain kind)",
+			vtxo:            domain.Vtxo{Kind: domain.VtxoKindOnchainPending},
+			requiresForfeit: false,
+		},
 	}
 	for _, f := range fixtures {
 		t.Run(f.name, func(t *testing.T) {

internal/infrastructure/db/badger/vtxo_repo.go

@@ -94,6 +94,55 @@ func (r *VtxoRepository) SpendVtxos(
 	return nil
 }
 
+// RecordCosignedTx implements domain.VtxoRepository.
+func (r *VtxoRepository) RecordCosignedTx(
+	ctx context.Context, txid string, inputs []domain.Outpoint, outputs []domain.Vtxo,
+) error {
+	pending := make([]domain.Vtxo, 0, len(outputs))
+	for _, out := range outputs {
+		out.Kind = domain.VtxoKindOnchainPending
+		pending = append(pending, out)
+	}
+
+	var err error
+	for range maxRetries {
+		err = func() error {
+			tx := r.store.Badger().NewTransaction(true)
+			defer tx.Discard()
+
+			// spendVtxo and addVtxos both pick a transaction up from the
+			// context, so the marks and the pending outputs commit together or
+			// not at all.
+			//nolint:staticcheck // the helpers read this key as a bare string
+			txCtx := context.WithValue(ctx, "tx", tx)
+
+			for _, in := range inputs {
+				// No ark txid. Its absence is what marks the spend as onchain
+				// rather than in-Ark.
+				if err := r.spendVtxo(txCtx, in, txid, ""); err != nil {
+					return err
+				}
+			}
+
+			if err := r.addVtxos(txCtx, pending); err != nil {
+				return err
+			}
+
+			return tx.Commit()
+		}()
+		if err == nil {
+			return nil
+		}
+		if errors.Is(err, badger.ErrConflict) {
+			time.Sleep(100 * time.Millisecond)
+			continue
+		}
+		return err
+	}
+
+	return err
+}
+
 func (r *VtxoRepository) UnrollVtxos(
 	ctx context.Context, outpoints []domain.Outpoint,
 ) error {

internal/infrastructure/db/cosigned_tx_repo_test.go

@@ -0,0 +1,122 @@
+package db_test
+
+import (
+	"context"
+	"testing"
+
+	"github.com/arkade-os/arkd/internal/core/domain"
+	"github.com/arkade-os/arkd/internal/infrastructure/db"
+	badgerdb "github.com/arkade-os/arkd/internal/infrastructure/db/badger"
+	"github.com/stretchr/testify/require"
+)
+
+// RecordCosignedTx is the durable write behind an on-chain cosign. The inputs
+// are marked spent with no ark txid, which is what tells an onchain spend from
+// an in-Ark one, and the outputs are recorded as pending because nothing arkd
+// has only cosigned is spendable yet.
+func TestRecordCosignedTx(t *testing.T) {
+	for name, repo := range cosignedTxRepos(t) {
+		t.Run(name, func(t *testing.T) {
+			ctx := context.Background()
+
+			t.Run("marks the inputs spent onchain and adds the outputs pending", func(t *testing.T) {
+				in := cosignInput(randomString(32))
+				require.NoError(t, repo.AddVtxos(ctx, []domain.Vtxo{in}))
+
+				spendTxid := randomString(32)
+				out := cosignOutput(spendTxid, 0)
+				require.NoError(t, repo.RecordCosignedTx(
+					ctx, spendTxid, []domain.Outpoint{in.Outpoint}, []domain.Vtxo{out},
+				))
+
+				got, err := repo.GetVtxos(ctx, []domain.Outpoint{in.Outpoint})
+				require.NoError(t, err)
+				require.Len(t, got, 1)
+				require.True(t, got[0].Spent)
+				require.Equal(t, spendTxid, got[0].SpentBy)
+				// The discriminator. An in-Ark spend would carry one of these.
+				require.Empty(t, got[0].ArkTxid)
+				require.Empty(t, got[0].SettledBy)
+
+				got, err = repo.GetVtxos(ctx, []domain.Outpoint{out.Outpoint})
+				require.NoError(t, err)
+				require.Len(t, got, 1)
+				require.Equal(t, domain.VtxoKindOnchainPending, got[0].Kind)
+				require.True(t, got[0].IsOnchainKind())
+				require.False(t, got[0].HasBatchExpiry())
+			})
+
+			// The caller cannot register something as already confirmed, since
+			// only the per-transaction lifecycle may promote a pending output.
+			t.Run("forces the pending kind whatever the caller asked for", func(t *testing.T) {
+				spendTxid := randomString(32)
+				out := cosignOutput(spendTxid, 1)
+				out.Kind = domain.VtxoKindOnchain
+
+				require.NoError(t, repo.RecordCosignedTx(ctx, spendTxid, nil, []domain.Vtxo{out}))
+
+				got, err := repo.GetVtxos(ctx, []domain.Outpoint{out.Outpoint})
+				require.NoError(t, err)
+				require.Len(t, got, 1)
+				require.Equal(t, domain.VtxoKindOnchainPending, got[0].Kind)
+			})
+
+			t.Run("an unknown input does not block the outputs", func(t *testing.T) {
+				spendTxid := randomString(32)
+				out := cosignOutput(spendTxid, 2)
+				missing := domain.Outpoint{Txid: randomString(32), VOut: 7}
+
+				require.NoError(t, repo.RecordCosignedTx(
+					ctx, spendTxid, []domain.Outpoint{missing}, []domain.Vtxo{out},
+				))
+
+				got, err := repo.GetVtxos(ctx, []domain.Outpoint{out.Outpoint})
+				require.NoError(t, err)
+				require.Len(t, got, 1)
+			})
+		})
+	}
+}
+
+func cosignedTxRepos(t *testing.T) map[string]domain.VtxoRepository {
+	t.Helper()
+
+	svc, err := db.NewService(db.ServiceConfig{
+		EventStoreType:   "badger",
+		DataStoreType:    "sqlite",
+		EventStoreConfig: []interface{}{"", nil},
+		DataStoreConfig:  []interface{}{t.TempDir()},
+		Settings:         validSettings(),
+	}, nil)
+	require.NoError(t, err)
+	t.Cleanup(svc.Close)
+
+	badgerRepo, err := badgerdb.NewVtxoRepository(t.TempDir(), nil)
+	require.NoError(t, err)
+	t.Cleanup(badgerRepo.Close)
+
+	return map[string]domain.VtxoRepository{
+		"sqlite": svc.Vtxos(),
+		"badger": badgerRepo,
+	}
+}
+
+func cosignInput(txid string) domain.Vtxo {
+	return domain.Vtxo{
+		Outpoint:           domain.Outpoint{Txid: txid, VOut: 0},
+		PubKey:             randomString(32),
+		Amount:             10_000,
+		RootCommitmentTxid: randomString(32),
+		CreatedAt:          1_700_000_000,
+		ExpiresAt:          2_000_000_000,
+	}
+}
+
+func cosignOutput(txid string, vout uint32) domain.Vtxo {
+	return domain.Vtxo{
+		Outpoint:  domain.Outpoint{Txid: txid, VOut: vout},
+		PubKey:    randomString(32),
+		Amount:    9_000,
+		CreatedAt: 1_700_000_000,
+	}
+}