iroh: working KV replication!

.vscode/settings.json

···
   },
   "mesonbuild.configureOnOpen": false,
   "cSpell.words": ["Devplace", "streamplace", "webrtc"],
-  "go.lintTool": "golangci-lint",
+  "go.lintTool": "golangci-lint-v2",
   "go.lintFlags": ["--path-mode=abs"],
   "go.formatTool": "custom",
   "go.alternateTools": {
-    "customFormatter": "golangci-lint"
+    "customFormatter": "golangci-lint-v2"
   },
   "go.formatFlags": ["fmt", "--stdin"],
   "editor.codeActionsOnSave": {

pkg/cmd/streamplace.go

···
 package cmd
 
 import (
+	"bytes"
 	"context"
 	"crypto"
+	"crypto/rand"
 	"errors"
 	"flag"
 	"fmt"
···
 	"stream.place/streamplace/pkg/notifications"
 	"stream.place/streamplace/pkg/replication"
 	"stream.place/streamplace/pkg/replication/boring"
+	"stream.place/streamplace/pkg/replication/iroh_replicator"
 	"stream.place/streamplace/pkg/rtmps"
 	v0 "stream.place/streamplace/pkg/schema/v0"
 	"stream.place/streamplace/pkg/spmetrics"
···
 		},
 	}
 
+	exists, err := cli.DataFileExists([]string{"iroh-kv-secret"})
+	if err != nil {
+		return err
+	}
+	if !exists {
+		secret := make([]byte, 32)
+		_, err := rand.Read(secret)
+		if err != nil {
+			return fmt.Errorf("failed to generate random secret: %w", err)
+		}
+		err = cli.DataFileWrite([]string{"iroh-kv-secret"}, bytes.NewReader(secret), true)
+		if err != nil {
+			return err
+		}
+	}
+	buf := bytes.Buffer{}
+	err = cli.DataFileRead([]string{"iroh-kv-secret"}, &buf)
+	if err != nil {
+		return err
+	}
+	secret := buf.Bytes()
+	swarm, err := iroh_replicator.StartKV(ctx, cli.Tickets, secret)
+	if err != nil {
+		return err
+	}
+
 	op := oatproxy.New(&oatproxy.Config{
 		Host:               cli.PublicHost,
 		CreateOAuthSession: state.CreateOAuthSession,
···
 		DownstreamJWK:      cli.AccessJWK,
 		ClientMetadata:     clientMetadata,
 	})
-	d := director.NewDirector(mm, mod, &cli, b, op, state)
+	d := director.NewDirector(mm, mod, &cli, b, op, state, swarm)
 	a, err := api.MakeStreamplaceAPI(&cli, mod, state, eip712signer, noter, mm, ms, b, atsync, d, op)
 	if err != nil {
 		return err
···
 
 	group.Go(func() error {
 		return mod.StartSegmentCleaner(ctx)
+	})
+
+	group.Go(func() error {
+		return swarm.Start(ctx, cli.Tickets)
 	})
 
 	if cli.LivepeerGateway {

pkg/config/config.go

···
 	SQLLogging             bool
 	SentryDSN              string
 	LivepeerDebug          bool
+	Tickets                []string
 }
 
 func (cli *CLI) NewFlagSet(name string) *flag.FlagSet {
···
 	fs.BoolVar(&cli.SQLLogging, "sql-logging", false, "enable sql logging")
 	fs.StringVar(&cli.SentryDSN, "sentry-dsn", "", "sentry dsn for error reporting")
 	fs.BoolVar(&cli.LivepeerDebug, "livepeer-debug", false, "log livepeer segments to $SP_DATA_DIR/livepeer-debug")
+	cli.StringSliceFlag(fs, &cli.Tickets, "tickets", "[]", "tickets to join the swarm with")
 
 	lpFlags := flag.NewFlagSet("livepeer", flag.ContinueOnError)
 	_ = starter.NewLivepeerConfig(lpFlags)

pkg/director/director.go

···
 	"fmt"
 	"sync"
 
+	"github.com/bluesky-social/indigo/util"
 	"github.com/streamplace/oatproxy/pkg/oatproxy"
 	"golang.org/x/sync/errgroup"
 	"stream.place/streamplace/pkg/bus"
···
 	"stream.place/streamplace/pkg/log"
 	"stream.place/streamplace/pkg/media"
 	"stream.place/streamplace/pkg/model"
+	"stream.place/streamplace/pkg/replication/iroh_replicator"
 	"stream.place/streamplace/pkg/statedb"
 )
 
···
 	streamSessionsMu sync.Mutex
 	op               *oatproxy.OATProxy
 	statefulDB       *statedb.StatefulDB
+	swarm            *iroh_replicator.SwarmKV
 }
 
-func NewDirector(mm *media.MediaManager, mod model.Model, cli *config.CLI, bus *bus.Bus, op *oatproxy.OATProxy, statefulDB *statedb.StatefulDB) *Director {
+func NewDirector(mm *media.MediaManager, mod model.Model, cli *config.CLI, bus *bus.Bus, op *oatproxy.OATProxy, statefulDB *statedb.StatefulDB, swarm *iroh_replicator.SwarmKV) *Director {
 	return &Director{
 		mm:               mm,
 		mod:              mod,
···
 		streamSessionsMu: sync.Mutex{},
 		op:               op,
 		statefulDB:       statefulDB,
+		swarm:            swarm,
 	}
 }
 
···
 				})
 			}
 			d.streamSessionsMu.Unlock()
+			go func() {
+				err := d.swarm.Put(ctx, not.Segment.RepoDID, not.Segment.StartTime.Format(util.ISO8601))
+				if err != nil {
+					log.Error(ctx, "could not put segment to swarm", "error", err)
+				}
+			}()
 			err := ss.NewSegment(ctx, not)
 			if err != nil {
 				log.Error(ctx, "could not add segment to stream session", "error", err)

pkg/iroh/generated/iroh_streamplace/iroh_streamplace.go

···
 		},
 	)
 
+	if err == nil {
+		return res, nil
+	}
+
 	return res, err
 }
 
···
 			C.ffi_iroh_streamplace_rust_future_free_pointer(handle)
 		},
 	)
+
+	if err == nil {
+		return res, nil
+	}
 
 	return res, err
 }
···
 		},
 	)
 
+	if err == nil {
+		return res, nil
+	}
+
 	return res, err
 }
 
···
 		},
 	)
 
+	if err == nil {
+		return res, nil
+	}
+
 	return res, err
 }
 
···
 			C.ffi_iroh_streamplace_rust_future_free_pointer(handle)
 		},
 	)
+
+	if err == nil {
+		return res, nil
+	}
 
 	return res, err
 }
···
 		},
 	)
 
+	if err == nil {
+		return nil
+	}
+
 	return err
 }
 
···
 			C.ffi_iroh_streamplace_rust_future_free_pointer(handle)
 		},
 	)
+
+	if err == nil {
+		return res, nil
+	}
 
 	return res, err
 }
···
 		},
 	)
 
+	if err == nil {
+		return nil
+	}
+
 	return err
 }
 
···
 		},
 	)
 
+	if err == nil {
+		return nil
+	}
+
 	return err
 }
 
···
 		},
 	)
 
+	if err == nil {
+		return res, nil
+	}
+
 	return res, err
 }
 
···
 			C.ffi_iroh_streamplace_rust_future_free_void(handle)
 		},
 	)
+
+	if err == nil {
+		return nil
+	}
 
 	return err
 }
···
 		},
 	)
 
+	if err == nil {
+		return res, nil
+	}
+
 	return res, err
 }
 
···
 		},
 	)
 
+	if err == nil {
+		return nil
+	}
+
 	return err
 }
 
···
 			C.ffi_iroh_streamplace_rust_future_free_void(handle)
 		},
 	)
+
+	if err == nil {
+		return nil
+	}
 
 	return err
 }
···
 			C.ffi_iroh_streamplace_rust_future_free_void(handle)
 		},
 	)
+
+	if err == nil {
+		return nil
+	}
 
 	return err
 }
···
 		},
 	)
 
+	if err == nil {
+		return res, nil
+	}
+
 	return res, err
 }
 
···
 			C.ffi_iroh_streamplace_rust_future_free_void(handle)
 		},
 	)
+
+	if err == nil {
+		return nil
+	}
 
 	return err
 }

pkg/replication/iroh/iroh.go pkg/replication/iroh_replicator/iroh.go

···
-package iroh
+package iroh_replicator
 
 import (
 	"context"

pkg/replication/iroh_replicator/kv.go

···
+package iroh_replicator
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	"stream.place/streamplace/pkg/iroh/generated/iroh_streamplace"
+	"stream.place/streamplace/pkg/log"
+)
+
+type SwarmKV struct {
+	node *iroh_streamplace.Node
+	db   *iroh_streamplace.Db
+	w    *iroh_streamplace.WriteScope
+}
+
+func StartKV(ctx context.Context, tickets []string, secret []byte) (*SwarmKV, error) {
+	ctx = log.WithLogValues(ctx, "func", "StartKV")
+
+	log.Log(ctx, "Starting with tickets", "tickets", tickets)
+	config := iroh_streamplace.Config{
+		Key:             secret,
+		Topic:           make([]byte, 32), // all zero topic for testing
+		MaxSendDuration: 1000_000_000,     // 1s
+	}
+	log.Log(ctx, "Config created", "config", config)
+	node, err := iroh_streamplace.NodeSender(config)
+	if err != nil {
+		return nil, fmt.Errorf("failed to create NodeSender: %w", err)
+	}
+
+	db := node.Db()
+	w := node.NodeScope()
+
+	node_id, err := node.NodeId()
+	if err != nil {
+		return nil, fmt.Errorf("failed to get NodeId: %w", err)
+	}
+	log.Log(ctx, "Node ID:", "node_id", node_id)
+
+	ticket, err := node.Ticket()
+	if err != nil {
+		return nil, fmt.Errorf("failed to get Ticket: %w", err)
+	}
+	log.Log(ctx, "Ticket:", "ticket", ticket)
+
+	swarm := SwarmKV{
+		node: node,
+		db:   db,
+		w:    w,
+	}
+	return &swarm, nil
+}
+
+func (swarm *SwarmKV) Start(ctx context.Context, tickets []string) error {
+	if len(tickets) > 0 {
+		err := swarm.node.JoinPeers(tickets)
+		if err != nil {
+			return fmt.Errorf("failed to join peers: %w", err)
+		}
+	}
+
+	sub := swarm.db.Subscribe(iroh_streamplace.NewFilter())
+	for {
+		if ctx.Err() != nil {
+			return ctx.Err()
+		}
+		ev, err := sub.NextRaw()
+		if err != nil {
+			return fmt.Errorf("failed to get next subscription event: %w", err)
+		}
+		if ev == nil {
+			log.Log(ctx, "Got empty event from sub.NextRaw(), pausing for a second")
+			time.Sleep(1 * time.Second)
+			continue
+		}
+		switch item := (*ev).(type) {
+		case iroh_streamplace.SubscribeItemEntry:
+			keyStr := string(item.Key)
+			valueStr := string(item.Value)
+			log.Log(ctx, "SubscribeItemEntry", "key", keyStr, "value", valueStr)
+
+		case iroh_streamplace.SubscribeItemCurrentDone:
+			log.Log(ctx, "SubscribeItemCurrentDone", "currentDone", item)
+		case iroh_streamplace.SubscribeItemExpired:
+			log.Log(ctx, "SubscribeItemExpired", "expired", item)
+		case iroh_streamplace.SubscribeItemOther:
+			log.Log(ctx, "SubscribeItemOther", "other", item)
+		}
+	}
+}
+
+func (swarm *SwarmKV) Put(ctx context.Context, key, value string) error {
+	// streamerBs := []byte(streamer)
+	keyBs := []byte(key)
+	valueBs := []byte(value)
+	return swarm.w.Put(nil, keyBs, valueBs)
+}

rust/iroh-streamplace/Makefile

···
-.PHONY: clean build-rust generate run
-
-install:
-	cargo install uniffi-bindgen-go --git https://github.com/NordSecurity/uniffi-bindgen-go --tag v0.4.0+v0.28.3
-
-clean:
-	cargo clean
-	rm -rf go/uniffi_example/
-
-build-rust:
-	cargo build
-
-generate: build-rust
-	uniffi-bindgen-go --library ../../target/debug/libiroh_streamplace.dylib --out-dir go/
-	cp ../../target/debug/libiroh_streamplace.dylib go/
-
-run: generate
-	cd go && CGO_LDFLAGS="-L. -liroh_streamplace" go run main.go $(TICKET)
-

rust/iroh-streamplace/go/go.mod

···
-module hello-app
-
-go 1.25.1

rust/iroh-streamplace/go/main.go

···
-
-package main
-
-import (
-	"fmt"
-	"os"
-	sp "hello-app/iroh_streamplace"
-	"reflect"
-    "crypto/rand"
-)
-
-func isNilError(err error) bool {
-    if err == nil {
-        return true
-    }
-    
-    v := reflect.ValueOf(err)
-    return v.Kind() == reflect.Ptr && v.IsNil()
-}
-
-
-func panicIfErr(err error) {
-	if !isNilError(err) {
-		panic(err)
-	}
-}
-
-func main() {
-	tickets := os.Args[1:];
-
-	secret := make([]byte, 32)
-    _, err := rand.Read(secret)
-    panicIfErr(err)
-
-	fmt.Println("Starting with tickets", tickets)
-	config := sp.Config {
-		Key : secret,
-		Topic: make([]byte, 32), // all zero topic for testing
-		MaxSendDuration: 1000_000_000, // 1s
-	}
-	fmt.Printf("Config created %+v\n", config)
-	node, err := sp.NodeSender(config)
-	panicIfErr(err)
-
-	db := node.Db()
-	w := node.NodeScope()
-
-	node_id, err := node.NodeId()
-	panicIfErr(err)
-	fmt.Println("Node ID:", node_id)
-
-	ticket, err := node.Ticket()
-	panicIfErr(err)
-	fmt.Println("Ticket:", ticket)
-
-	if len(tickets) > 0 {
-		err = node.JoinPeers(tickets)
-		panicIfErr(err)
-	}
-
-	w.Put(nil, []byte("hello"), []byte("world"))
-	stream := []byte("stream1")
-	w.Put(&stream, []byte("subscribed"), []byte("true"))
-
-	filter := sp.NewFilter()
-	items, err := db.IterWithOpts(filter)
-	panicIfErr(err)
-	fmt.Printf("Iter items: %+v\n", items)
-
-	filter2 := sp.NewFilter().Global()
-	items2, err := db.IterWithOpts(filter2)
-	panicIfErr(err)
-	fmt.Printf("Iter items: %+v\n", items2)
-
-	filter3 := sp.NewFilter().Stream(stream)
-	items3, err := db.IterWithOpts(filter3)
-	panicIfErr(err)
-	fmt.Printf("Iter items: %+v\n", items3)
-
-	sub := db.Subscribe(sp.NewFilter())
-	for {
-		ev, err := sub.NextRaw()
-		panicIfErr(err)
-		switch (*ev).(type) {
-			case sp.SubscribeItemEntry:
-				fmt.Printf("%+v\n", (*ev).(sp.SubscribeItemEntry))
-			case sp.SubscribeItemCurrentDone:
-				fmt.Printf("Got current done event: %+v\n", (*ev).(sp.SubscribeItemCurrentDone))
-			case sp.SubscribeItemExpired:
-				fmt.Printf("Got expired event: %+v\n", (*ev).(sp.SubscribeItemExpired))
-			case sp.SubscribeItemOther:
-				fmt.Printf("Got other event: %+v\n", (*ev).(sp.SubscribeItemOther))
-		}
-	}
-}