iroh replicator: extremely rough PoC

pkg/cmd/streamplace.go

···
 	"stream.place/streamplace/pkg/log"
 	"stream.place/streamplace/pkg/media"
 	"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"
···
 		log.Log(ctx, "successfully initialized hardware signer", "address", addr)
 		signer = hwsigner
 	}
-	var rep replication.Replicator = &boring.BoringReplicator{Peers: cli.Peers}
+	// var rep replication.Replicator = &boring.BoringReplicator{Peers: cli.Peers}
 
 	mod, err := model.MakeDB(cli.DataFilePath([]string{"index"}))
 	if err != nil {

pkg/director/director.go

···
 
 import (
 	"context"
+	"encoding/json"
 	"fmt"
 	"sync"
 
···
 }
 
 func (d *Director) Start(ctx context.Context) error {
+	nodeId, err := d.swarm.Node.NodeId()
+	if err != nil {
+		return fmt.Errorf("failed to get node id: %w", err)
+	}
+
 	newSeg := d.mm.NewSegment()
 	ctx, cancel := context.WithCancel(ctx)
 	defer cancel()
···
 			}
 			d.streamSessionsMu.Unlock()
 			go func() {
-				err := d.swarm.Put(ctx, not.Segment.RepoDID, not.Segment.StartTime.Format(util.ISO8601))
+				originInfo := iroh_replicator.OriginInfo{
+					NodeID: nodeId.String(),
+					Time:   not.Segment.StartTime.Format(util.ISO8601),
+				}
+				bs, err := json.Marshal(originInfo)
+				if err != nil {
+					log.Error(ctx, "could not marshal origin info", "error", err)
+					return
+				}
+				err = d.swarm.Put(ctx, not.Segment.RepoDID, bs)
 				if err != nil {
 					log.Error(ctx, "could not put segment to swarm", "error", err)
+					return
+				}
+				err = d.swarm.Node.SendSegment(not.Segment.RepoDID, not.Data)
+				if err != nil {
+					log.Error(ctx, "could not send segment to swarm", "error", err)
+					return
 				}
 			}()
 			err := ss.NewSegment(ctx, not)

pkg/replication/iroh_replicator/kv.go

···
 
 import (
 	"context"
+	"encoding/json"
 	"fmt"
 	"time"
 
···
 )
 
 type SwarmKV struct {
-	node *iroh_streamplace.Node
-	db   *iroh_streamplace.Db
+	Node *iroh_streamplace.Node
+	DB   *iroh_streamplace.Db
 	w    *iroh_streamplace.WriteScope
+}
+
+// A message saying "hey I ingested node data at this time"
+type OriginInfo struct {
+	NodeID string `json:"node_id"`
+	Time   string `json:"time"`
+}
+
+type DataHandler struct{}
+
+func (handler *DataHandler) HandleData(topic string, data []byte) {
+	log.Log(context.Background(), "HandleData", "topic", topic, "data", len(data))
 }
 
 func StartKV(ctx context.Context, tickets []string, secret []byte) (*SwarmKV, error) {
+	handler := &DataHandler{}
 	ctx = log.WithLogValues(ctx, "func", "StartKV")
 
 	log.Log(ctx, "Starting with tickets", "tickets", tickets)
···
 		MaxSendDuration: 1000_000_000,     // 1s
 	}
 	log.Log(ctx, "Config created", "config", config)
-	node, err := iroh_streamplace.NodeSender(config)
+	node, err := iroh_streamplace.NodeReceiver(config, handler)
 	if err != nil {
 		return nil, fmt.Errorf("failed to create NodeSender: %w", err)
 	}
···
 	log.Log(ctx, "Ticket:", "ticket", ticket)
 
 	swarm := SwarmKV{
-		node: node,
-		db:   db,
+		Node: node,
+		DB:   db,
 		w:    w,
 	}
 	return &swarm, nil
 }
 
+var activeSubs = make(map[string]bool)
+
 func (swarm *SwarmKV) Start(ctx context.Context, tickets []string) error {
 	if len(tickets) > 0 {
-		err := swarm.node.JoinPeers(tickets)
+		err := swarm.Node.JoinPeers(tickets)
 		if err != nil {
 			return fmt.Errorf("failed to join peers: %w", err)
 		}
 	}
 
-	sub := swarm.db.Subscribe(iroh_streamplace.NewFilter())
+	nodeId, err := swarm.Node.NodeId()
+	if err != nil {
+		return fmt.Errorf("failed to get node id: %w", err)
+	}
+	nodeIdStr := nodeId.String()
+	log.Log(ctx, "Node ID:", "node_id", nodeIdStr)
+
+	sub := swarm.DB.Subscribe(iroh_streamplace.NewFilter())
 	for {
 		if ctx.Err() != nil {
 			return ctx.Err()
···
 			keyStr := string(item.Key)
 			valueStr := string(item.Value)
 			log.Log(ctx, "SubscribeItemEntry", "key", keyStr, "value", valueStr)
+			var info OriginInfo
+			err := json.Unmarshal(item.Value, &info)
+			if err != nil {
+				log.Error(ctx, "could not unmarshal origin info", "error", err)
+				continue
+			}
+			if !activeSubs[keyStr] {
+				if info.NodeID == nodeIdStr {
+					activeSubs[keyStr] = true
+					continue
+				}
+				pubKey, err := iroh_streamplace.PublicKeyFromString(info.NodeID)
+				if err != nil {
+					log.Error(ctx, "could not create public key", "error", err)
+					continue
+				}
+				activeSubs[keyStr] = true
+				err = swarm.Node.Subscribe(keyStr, pubKey)
+				if err != nil {
+					log.Error(ctx, "could not subscribe to key", "error", err)
+					continue
+				}
+			}
 
 		case iroh_streamplace.SubscribeItemCurrentDone:
 			log.Log(ctx, "SubscribeItemCurrentDone", "currentDone", item)
···
 	}
 }
 
-func (swarm *SwarmKV) Put(ctx context.Context, key, value string) error {
+func (swarm *SwarmKV) Put(ctx context.Context, key string, value []byte) error {
 	// streamerBs := []byte(streamer)
 	keyBs := []byte(key)
-	valueBs := []byte(value)
-	return swarm.w.Put(nil, keyBs, valueBs)
+	return swarm.w.Put(nil, keyBs, value)
 }