diff --git a/go.mod b/go.mod index aaf13b9a..894d16cf 100644 --- a/go.mod +++ b/go.mod @@ -306,6 +306,7 @@ require ( github.com/multiformats/go-multibase v0.2.0 // indirect github.com/multiformats/go-varint v0.0.7 // indirect github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect + github.com/n0-computer/iroh-streamplace v0.0.0-20250805111327-449fe3719c26 // indirect github.com/nakabonne/nestif v0.3.1 // indirect github.com/nishanths/exhaustive v0.12.0 // indirect github.com/nishanths/predeclared v0.2.2 // indirect diff --git a/go.sum b/go.sum index e44a2eb4..b915ff7d 100644 --- a/go.sum +++ b/go.sum @@ -737,6 +737,8 @@ github.com/multiformats/go-varint v0.0.7 h1:sWSGR+f/eu5ABZA2ZpYKBILXTTs9JWpdEM/n github.com/multiformats/go-varint v0.0.7/go.mod h1:r8PUYw/fD/SjBCiKOoDlGF6QawOELpZAu9eioSos/OU= github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 h1:C3w9PqII01/Oq1c1nUAm88MOHcQC9l5mIlSMApZMrHA= github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822/go.mod h1:+n7T8mK8HuQTcFwEeznm/DIxMOiR9yIdICNftLE1DvQ= +github.com/n0-computer/iroh-streamplace v0.0.0-20250805111327-449fe3719c26 h1:QUZ2LCu5a8mRhGWZHS5hnZiEiVXq6uQP8RZdSSPJ140= +github.com/n0-computer/iroh-streamplace v0.0.0-20250805111327-449fe3719c26/go.mod h1:rDqM/Vg0c1Vd49Uztot7HQ5VYghhKkf7tZZegO1smlM= github.com/nakabonne/nestif v0.3.1 h1:wm28nZjhQY5HyYPx+weN3Q65k6ilSBxDb8v5S81B81U= github.com/nakabonne/nestif v0.3.1/go.mod h1:9EtoZochLn5iUprVDmDjqGKPofoUEBL8U4Ngq6aY7OE= github.com/nishanths/exhaustive v0.12.0 h1:vIY9sALmw6T/yxiASewa4TQcFsVYZQQRUQJhKRf3Swg= diff --git a/pkg/cmd/streamplace.go b/pkg/cmd/streamplace.go index 8ec66040..fbce2b30 100644 --- a/pkg/cmd/streamplace.go +++ b/pkg/cmd/streamplace.go @@ -28,7 +28,7 @@ import ( "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" "stream.place/streamplace/pkg/resync" "stream.place/streamplace/pkg/rtmps" v0 "stream.place/streamplace/pkg/schema/v0" @@ -266,7 +266,8 @@ func start(build *config.BuildFlags, platformJobs []jobFunc) error { 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} + var rep replication.Replicator = iroh.NewIrohReplicator(cli.Peers) mod, err := model.MakeDB(cli.DBPath) if err != nil { return err diff --git a/pkg/replication/iroh/iroh.go b/pkg/replication/iroh/iroh.go new file mode 100644 index 00000000..85f58d1e --- /dev/null +++ b/pkg/replication/iroh/iroh.go @@ -0,0 +1,48 @@ +package iroh + +import ( + "context" + + "stream.place/streamplace/pkg/log" + + irohStreamplace "github.com/n0-computer/iroh-streamplace/pkg/iroh_streamplace/generated/iroh_streamplace" +) + +// iroh replication mechanism +type IrohReplicator struct { + peers []*irohStreamplace.PublicKey + endpoint *irohStreamplace.SenderEndpoint +} + +func NewIrohReplicator(peers []string) *IrohReplicator { + endpoint := irohStreamplace.NewSenderEndpoint() + // TODO: add addrs to the endpoint + + nodeAddrs := make([]*irohStreamplace.PublicKey, len(peers)) + for i := range(peers) { + addr := irohStreamplace.PublicKeyFromString(peers[i]) + nodeAddrs[i] = addr + } + + return &IrohReplicator { + peers: nodeAddrs, + endpoint: endpoint, + } +} + +func (rep *IrohReplicator) NewSegment(ctx context.Context, bs []byte) { + for _, p := range rep.peers { + go func(peer *irohStreamplace.PublicKey) { + ctx := log.WithLogValues(ctx, "peer", peer.FmtShort()) + err := sendSegment(ctx, rep.endpoint, peer, bs) + if err != nil { + log.Log(ctx, "error replicating segment", "error", err) + } + }(p) + } +} + +func sendSegment(ctx context.Context, endpoint *irohStreamplace.SenderEndpoint, peer *irohStreamplace.PublicKey, bs []byte) error { + endpoint.Send(peer, bs) + return nil +} diff --git a/subprojects/.gitignore b/subprojects/.gitignore index ee2eeac0..760cb32b 100644 --- a/subprojects/.gitignore +++ b/subprojects/.gitignore @@ -7,3 +7,4 @@ !atproto.wrap !packagefiles/ !packagefiles/** +!iroh_streamplace.wrap diff --git a/subprojects/iroh_streamplace.wrap b/subprojects/iroh_streamplace.wrap new file mode 100644 index 00000000..629f0e54 --- /dev/null +++ b/subprojects/iroh_streamplace.wrap @@ -0,0 +1,4 @@ +[wrap-git] +url = https://github.com/n0-computer/iroh-streamplace.git +revision = v0.1.0 +depth = 1