1package replication23import (4 "context"5 "encoding/json"6 "errors"7 "fmt"8 "net/url"9 "strings"1011 "github.com/FelineStateMachine/tinyrelay/internal/event"12 "github.com/FelineStateMachine/tinyrelay/internal/storage"13)1415// Origin identifies how an event entered the relay and controls whether it is16// eligible for outbound fanout.17type Origin string1819const (20 OriginClient Origin = "client"21 OriginImport Origin = "import"22 OriginServer Origin = "server"23 OriginPush Origin = "push"24 OriginBackupRestore Origin = "backup_restore"25)2627// Policy supplies current fanout decisions and relay identity. The host28// controls process capacity independently.29type Policy struct {30 Enabled bool31 ReadMembersOnly bool32 SelfPubKey string33 SelfRelayURL string34}3536// RelayDirectory returns write relays for an author and read relays for a37// recipient. Returned URLs are filtered by PrepareIntents.38type RelayDirectory interface {39 WriteRelays(pubkey string) []string40 ReadRelays(pubkey string) []string41}4243// RelayListDiscovery returns the source-ordered NIP-65 relay list used by44// owner backfill. RelayDirectory's directional methods remain sufficient for45// delivery planning.46type RelayListDiscovery interface {47 RelayList(pubkey string) []string48}4950// Discovery resolves write relays for an author.51type Discovery interface {52 DiscoverRelays(ctx context.Context, pubkey string) ([]string, error)53}5455// ReadDiscovery resolves read relays for a recipient.56type ReadDiscovery interface {57 DiscoverReadRelays(ctx context.Context, pubkey string) ([]string, error)58}5960// IntentPayload records the event origin in a delivery intent.61type IntentPayload struct {62 Origin Origin `json:"origin"`63}6465// PrepareIntents computes NIP-65 outbox work. The event transaction should66// pass the returned references to storage.Save as SaveOptions.Intents.67func PrepareIntents(ctx context.Context, e event.Event, origin Origin, policy Policy, directory RelayDirectory) []storage.Intent {68 if err := contextErr(ctx); err != nil || !eligible(e, origin, policy) {69 return nil70 }71 targets := uniqueTargets(directory.WriteRelays(e.PubKey))72 for _, pubkey := range event.TagValues(e, "p") {73 targets = appendUnique(targets, directory.ReadRelays(pubkey))74 }75 payload, err := json.Marshal(IntentPayload{Origin: origin})76 if err != nil {77 return nil78 }79 intents := make([]storage.Intent, 0, len(targets))80 for _, target := range targets {81 if !safeRelayURL(target) {82 continue83 }84 if policy.SelfRelayURL != "" && canonicalRelayURL(target) == canonicalRelayURL(policy.SelfRelayURL) {85 continue86 }87 intents = append(intents, storage.Intent{Kind: "delivery", EventID: e.ID, Target: target, Payload: string(payload)})88 }89 return intents90}9192func contextErr(ctx context.Context) error {93 select {94 case <-ctx.Done():95 return ctx.Err()96 default:97 return nil98 }99}100101func eligible(e event.Event, origin Origin, policy Policy) bool {102 peerPublication := origin == OriginServer && e.Kind == event.KIND_RELAY_DISCOVERY && e.PubKey == policy.SelfPubKey103 if !policy.Enabled || policy.ReadMembersOnly || (origin != OriginClient && !peerPublication) || (e.PubKey == policy.SelfPubKey && !peerPublication) || privateKind(e.Kind) {104 return false105 }106 for _, tag := range e.Tags {107 if len(tag) > 0 && tag[0] == "-" {108 return false109 }110 }111 return true112}113114func privateKind(kind int) bool {115 return event.IsPrivate(kind) || kind == event.KIND_DM116}117118func uniqueTargets(values []string) []string {119 out := make([]string, 0, len(values))120 for _, value := range values {121 if safeRelayURL(value) {122 out = appendUnique(out, []string{value})123 }124 }125 return out126}127128func appendUnique(out []string, values []string) []string {129 for _, value := range values {130 value = strings.TrimRight(strings.TrimSpace(value), "/")131 if value == "" || contains(out, value) {132 continue133 }134 out = append(out, value)135 }136 return out137}138139func contains(values []string, wanted string) bool {140 for _, value := range values {141 if value == wanted {142 return true143 }144 }145 return false146}147148func safeRelayURL(raw string) bool {149 u, err := url.Parse(strings.TrimRight(strings.TrimSpace(raw), "/"))150 if err != nil || (u.Scheme != "ws" && u.Scheme != "wss") || u.Host == "" || u.User != nil || u.Fragment != "" {151 return false152 }153 return true154}155156func canonicalRelayURL(raw string) string {157 u, err := url.Parse(strings.TrimSpace(raw))158 if err != nil || u.Host == "" {159 return strings.TrimRight(strings.TrimSpace(raw), "/")160 }161 scheme := strings.ToLower(u.Scheme)162 switch scheme {163 case "ws":164 scheme = "http"165 case "wss":166 scheme = "https"167 }168 host := strings.ToLower(u.Hostname())169 if port := u.Port(); port != "" {170 host += ":" + port171 }172 path := strings.TrimRight(u.EscapedPath(), "/")173 if path == "" {174 path = "/"175 }176 return scheme + "://" + host + path + "?" + u.RawQuery177}178179// DeliveryTransport sends one event to a relay target. The implementation180// owns connection reuse, request deadlines and transport shutdown.181type DeliveryTransport interface {182 Send(ctx context.Context, target string, e event.Event) (DeliveryResult, error)183}184185// DeliveryResult reports the relay's acceptance response. Duplicate186// acknowledgments are treated as successful delivery by the handlers.187type DeliveryResult struct {188 Accepted bool189 Message string190}191192func ValidateIntent(intent storage.Intent) error {193 if intent.Kind == "" || intent.EventID == "" || !safeRelayURL(intent.Target) {194 return errors.New("replication: malformed intent")195 }196 return nil197}198199func DecodeIntentPayload(raw string) (IntentPayload, error) {200 var payload IntentPayload201 if err := json.Unmarshal([]byte(raw), &payload); err != nil {202 return payload, fmt.Errorf("replication: decode intent payload: %w", err)203 }204 return payload, nil205}