package app import ( "bytes" "context" "encoding/hex" "encoding/json" "fmt" "lol.mleku.dev/chk" "lol.mleku.dev/log" "git.mleku.dev/mleku/nostr/encoders/filter" "git.mleku.dev/mleku/nostr/encoders/tag" commonv1 "next.orly.dev/pkg/proto/orlysync/common/v1" negentropygrpc "next.orly.dev/pkg/sync/negentropy/grpc" ) // NIP-77 Negentropy envelope constants const ( NegOpenLabel = "NEG-OPEN" NegMsgLabel = "NEG-MSG" NegCloseLabel = "NEG-CLOSE" NegErrLabel = "NEG-ERR" ) // negentropyClient is the gRPC client for negentropy operations // This should be set via Server.SetNegentropyClient() when using gRPC mode var negentropyClient *negentropygrpc.Client // SetNegentropyClient sets the negentropy gRPC client for NIP-77 WebSocket handling func SetNegentropyClient(client *negentropygrpc.Client) { negentropyClient = client } // IsNegentropyEnvelope checks if a message starts with a NEG-* envelope type func IsNegentropyEnvelope(msg []byte) bool { // Quick check: must start with '["NEG-' if len(msg) < 8 { return false } return bytes.HasPrefix(msg, []byte(`["NEG-`)) } // IdentifyNegentropyEnvelope extracts the envelope type and remaining payload func IdentifyNegentropyEnvelope(msg []byte) (envelopeType string, ok bool) { // Parse enough to get the envelope type if !IsNegentropyEnvelope(msg) { return "", false } // Find the first comma after the opening label end := bytes.IndexByte(msg[2:], '"') if end < 0 { return "", false } envelopeType = string(msg[2 : 2+end]) return envelopeType, true } // HandleNegOpen processes NEG-OPEN messages // Format: ["NEG-OPEN", subscription_id, filter, initial_message?] func (l *Listener) HandleNegOpen(msg []byte) error { log.I.F("HandleNegOpen called from %s", l.connectionID) if negentropyClient == nil { log.E.F("negentropy client is nil") return l.sendNegErr("", "negentropy not enabled") } // Parse the message array var parts []json.RawMessage if err := json.Unmarshal(msg, &parts); err != nil { return l.sendNegErr("", fmt.Sprintf("invalid NEG-OPEN format: %v", err)) } if len(parts) < 3 { return l.sendNegErr("", "NEG-OPEN requires at least 3 elements") } // Extract subscription ID var subscriptionID string if err := json.Unmarshal(parts[1], &subscriptionID); err != nil { return l.sendNegErr("", fmt.Sprintf("invalid subscription_id: %v", err)) } // Extract filter var f filter.F if err := json.Unmarshal(parts[2], &f); err != nil { return l.sendNegErr(subscriptionID, fmt.Sprintf("invalid filter: %v", err)) } // Extract optional initial message (hex encoded per NIP-77) var initialMessage []byte if len(parts) >= 4 { var msgStr string if err := json.Unmarshal(parts[3], &msgStr); err == nil && msgStr != "" { // NIP-77 uses hex encoding if decoded, err := hex.DecodeString(msgStr); err == nil { initialMessage = decoded } else { log.W.F("NEG-OPEN: invalid hex message: %v", err) } } } // Convert filter to proto format protoFilter := filterToProto(&f) // Call gRPC service ctx := context.Background() respMsg, errStr, err := negentropyClient.HandleNegOpen( ctx, l.connectionID, subscriptionID, protoFilter, initialMessage, ) if err != nil { log.E.F("NEG-OPEN gRPC error: %v", err) return l.sendNegErr(subscriptionID, "internal error") } if errStr != "" { return l.sendNegErr(subscriptionID, errStr) } // Send NEG-MSG response with initial message return l.sendNegMsg(subscriptionID, respMsg) } // HandleNegMsg processes NEG-MSG messages // Format: ["NEG-MSG", subscription_id, message] func (l *Listener) HandleNegMsg(msg []byte) error { if negentropyClient == nil { return l.sendNegErr("", "negentropy not enabled") } // Parse the message array var parts []json.RawMessage if err := json.Unmarshal(msg, &parts); err != nil { return l.sendNegErr("", fmt.Sprintf("invalid NEG-MSG format: %v", err)) } if len(parts) < 3 { return l.sendNegErr("", "NEG-MSG requires 3 elements") } // Extract subscription ID var subscriptionID string if err := json.Unmarshal(parts[1], &subscriptionID); err != nil { return l.sendNegErr("", fmt.Sprintf("invalid subscription_id: %v", err)) } // Extract message (hex or base64 encoded) var msgStr string if err := json.Unmarshal(parts[2], &msgStr); err != nil { return l.sendNegErr(subscriptionID, fmt.Sprintf("invalid message: %v", err)) } // Decode message (NIP-77 uses hex encoding) negentropyMsg, err := hex.DecodeString(msgStr) if err != nil { return l.sendNegErr(subscriptionID, fmt.Sprintf("invalid hex message: %v", err)) } // Call gRPC service ctx := context.Background() respMsg, haveIDs, needIDs, complete, errStr, err := negentropyClient.HandleNegMsg( ctx, l.connectionID, subscriptionID, negentropyMsg, ) if err != nil { log.E.F("NEG-MSG gRPC error: %v", err) return l.sendNegErr(subscriptionID, "internal error") } if errStr != "" { return l.sendNegErr(subscriptionID, errStr) } // If we have events to send (have_ids), fetch and send them if len(haveIDs) > 0 { if err := l.sendEventsForIDs(subscriptionID, haveIDs); err != nil { log.E.F("failed to send events for NEG-MSG: %v", err) } } // Log need_ids (events client should send us) if len(needIDs) > 0 { log.D.F("NEG-MSG: client needs to send %d events", len(needIDs)) } // Send NEG-MSG response if err := l.sendNegMsg(subscriptionID, respMsg); err != nil { return err } // If reconciliation is complete, log it if complete { log.D.F("NEG-MSG: reconciliation complete for %s", subscriptionID) } return nil } // HandleNegClose processes NEG-CLOSE messages // Format: ["NEG-CLOSE", subscription_id] func (l *Listener) HandleNegClose(msg []byte) error { if negentropyClient == nil { return nil // Silently ignore if not enabled } // Parse the message array var parts []json.RawMessage if err := json.Unmarshal(msg, &parts); err != nil { return nil // Silently ignore malformed close } if len(parts) < 2 { return nil } // Extract subscription ID var subscriptionID string if err := json.Unmarshal(parts[1], &subscriptionID); err != nil { return nil } // Call gRPC service to close the session ctx := context.Background() if err := negentropyClient.HandleNegClose(ctx, l.connectionID, subscriptionID); err != nil { log.E.F("NEG-CLOSE gRPC error: %v", err) } return nil } // sendNegMsg sends a NEG-MSG response to the client func (l *Listener) sendNegMsg(subscriptionID string, message []byte) error { // Encode message as hex (per NIP-77) encoded := "" if len(message) > 0 { encoded = hex.EncodeToString(message) } // Format: ["NEG-MSG", subscription_id, message] resp, err := json.Marshal([]any{NegMsgLabel, subscriptionID, encoded}) if err != nil { return err } _, err = l.Write(resp) return err } // sendNegErr sends a NEG-ERR response to the client func (l *Listener) sendNegErr(subscriptionID, reason string) error { // Format: ["NEG-ERR", subscription_id, reason] resp, err := json.Marshal([]any{NegErrLabel, subscriptionID, reason}) if err != nil { return err } _, err = l.Write(resp) return err } // sendEventsForIDs fetches and sends events for the given IDs func (l *Listener) sendEventsForIDs(subscriptionID string, ids [][]byte) error { if len(ids) == 0 { return nil } log.I.F("NEG: sending %d events for subscription %s", len(ids), subscriptionID) // Build filter with hex-encoded IDs f := &filter.F{} f.Ids = &tag.T{} for _, id := range ids { // IDs can be 32 bytes (full) or 16 bytes (truncated per NIP-77) hexID := hex.EncodeToString(id) f.Ids.T = append(f.Ids.T, []byte(hexID)) } // Query events by IDs ctx := l.ctx events, err := l.Server.db.QueryEvents(ctx, f) if err != nil { log.E.F("NEG: failed to query events: %v", err) return err } // Send each event via EVENT envelope sent := 0 for _, ev := range events { if ev == nil { continue } if err := l.SendEvent(ev); err != nil { log.W.F("NEG: failed to send event: %v", err) continue } sent++ } log.I.F("NEG: sent %d/%d events for subscription %s", sent, len(ids), subscriptionID) return nil } // filterToProto converts a nostr filter to proto format func filterToProto(f *filter.F) *commonv1.Filter { if f == nil { return nil } pf := &commonv1.Filter{} // Convert Ids if f.Ids != nil { for _, id := range f.Ids.T { pf.Ids = append(pf.Ids, id) } } // Convert Authors if f.Authors != nil { for _, author := range f.Authors.T { pf.Authors = append(pf.Authors, author) } } // Convert Kinds - kind.S has ToUint16() method if f.Kinds != nil && f.Kinds.Len() > 0 { for _, k := range f.Kinds.ToUint16() { pf.Kinds = append(pf.Kinds, uint32(k)) } } // Convert Since/Until - timestamp.T has .V field (int64) if f.Since != nil && f.Since.V != 0 { since := f.Since.V pf.Since = &since } if f.Until != nil && f.Until.V != 0 { until := f.Until.V pf.Until = &until } // Convert Limit if f.Limit != nil { limit := uint32(*f.Limit) pf.Limit = &limit } // Note: Tag filters (e, p, etc.) would need more complex conversion // This is a simplified implementation return pf } // CloseAllNegentropySessions closes all negentropy sessions for a connection // Called when a WebSocket connection is closed func (l *Listener) CloseAllNegentropySessions() { if negentropyClient == nil { return } ctx := context.Background() sessions, err := negentropyClient.ListSessions(ctx) if chk.E(err) { return } for _, sess := range sessions { if sess.ConnectionID == l.connectionID { negentropyClient.CloseSession(ctx, l.connectionID, sess.SubscriptionID) } } }