package main

import (
	"runtime"
	"git.smesh.lol/moxie/pkg/mxutil"
	"git.smesh.lol/musiquay/web/common/helpers"
	"git.smesh.lol/musiquay/web/common/jsbridge/relayproxy"
	"git.smesh.lol/musiquay/web/common/jsbridge/ws"
	"git.smesh.lol/musiquay/web/common/mw"
	"git.smesh.lol/nostr/pkg/core"
	"git.smesh.lol/musiquay/web/common/relay"
)

// Relay-proxy worker.
//
// Owns: WS pool to remote nostr relays, subscription router, MLS event
// subscription routing (kinds 443/445/1059 -> MLS worker via supervisor).
//
// IDB is owned by the Store Worker. This worker sends S_* messages to the
// supervisor which routes them to the Store Worker, and receives SR_* responses.
//
// Wire format: JSON-array MW strings.
//   page -> worker:
//     ["REQ", subID, filter]
//     ["CLOSE", subID]
//     ["PROXY", subID, filter, [relayURLs]]
//     ["EVENT", signedEventJSON]                  -- publish to ps.writeRelays
//     ["PUBLISH_TO", signedEventJSON, [relayURLs]] -- explicit targets
//     ["SET_PUBKEY", hex]
//     ["SET_WRITE_RELAYS", [relayURLs]]
//     ["ENC_KEY", hex]
//     ["MLS_SUB", [relayURLs], [groupIDs]]  -- open persistent MLS subs
//     ["MLS_UPDATE_GROUPS", [groupIDs]]     -- update kind 445 #h filter
//     ["MLS_FETCH_KP", peer, [relayURLs]]  -- one-shot key package fetch
//     ["SR_QUERY", reqID, eventsJSON]       -- response from Store Worker
//   worker -> page:
//     ["READY"]
//     ["EVENT", subID, event]
//     ["EOSE", subID]
//     ["OK", eventID, ok, message]
//     ["MLS_EVENT", evJSON]                 -- kind 443/445/1059 -> MLS worker
//     ["MLS_KP_RESULT", peer, evJSON]       -- one-shot KP fetch result -> MLS worker
//     ["S_PUT_EVENT", evJSON]               -- to Store Worker (fire-and-forget)
//     ["S_ENC_KEY", hexKey]                 -- to Store Worker (fire-and-forget)
//     ["S_QUERY", reqID, filterJSON]        -- to Store Worker

type peerRelayInfo struct {
	urls []string // relays the peer reads from (or both read+write)
	ts   int64    // created_at of the kind 10002 event
}

type clientSub struct {
	filter    *nostr.Filter
	filterRaw string
}

type proxySub struct {
	remoteIDs  map[string]bool
	relayCount int32
	timer      int32
	done       bool
	live       bool
	// Initial load ends when BOTH sources are done: the local store query and
	// the relays (or the fallback timer). Whichever finishes last decides, so
	// events from the slower source are rendered as part of the initial load
	// instead of being buffered as new posts.
	localDone  bool
	remoteDone bool
}

// proxyState is the relay-proxy worker's mutable state. Package globals are
// immutable outside initialization, so its timers, subscription maps and
// in-flight fetch bookkeeping live in one self-mutating type reached through
// a package-level pointer.
type proxyState struct {
	// eoseTimeoutMs: how long to wait for relays to EOSE before emitting EOSE
	// to the consumer. Short value gets the consumer unblocked quickly so it
	// can render cached data; the linger window catches stragglers.
	eoseTimeoutMs int32
	// lingerMs: after EOSE is emitted to the consumer, keep the remote subs
	// open this long to forward late-arriving events. Solves the slow-relay
	// race where events arrived just after the EOSE timeout fired.
	lingerMs int32

	clientSubs    map[string]*clientSub
	proxySubs     map[string]*proxySub
	remoteToProxy map[string]string
	rpool         *relay.Pool
	writeRelays   []string
	myPubkey      string

	mlsSubIDs      map[string]bool // MLS event subscriptions (kinds 443/445/1059)
	mlsKPFetchID   string          // in-flight one-shot KP fetch prefix
	mlsKPFetchPeer string          // peer pubkey for in-flight KP fetch
	mlsKPFetchSubs []string        // per-relay sub IDs for in-flight fetch
	mlsGroupIDs    []string        // current #h filter for kind 445

	storeCBs    map[int32]func(string) // reqID -> callback for SR_QUERY responses
	nextStoreID int32

	// Peer relay list cache (kind 10002).
	peerRelayCache map[string]*peerRelayInfo // pubkey -> cached relay list
	rlFetchID      string                    // in-flight relay list fetch sub prefix
	rlFetchPeer    string                    // peer for in-flight relay list fetch
	rlFetchSubs    []string                  // per-relay sub IDs
	rlFetchCB      func([]string)            // callback when relay list is resolved
	rlFetchEOSE    int32                     // count of EOSE received
}

var ps *proxyState

func initState() {
	if ps != nil {
		return
	}
	ps = &proxyState{}
	ps.eoseTimeoutMs = 3000
	ps.lingerMs = 25000
	ps.clientSubs = map[string]*clientSub{}
	ps.proxySubs = map[string]*proxySub{}
	ps.remoteToProxy = map[string]string{}
	ps.rpool = relay.NewPool()
	ps.mlsSubIDs = map[string]bool{}
	ps.storeCBs = map[int32]func(string){}
	ps.peerRelayCache = map[string]*peerRelayInfo{}
}

func main() {
	initState()
	relayproxy.WorkerOnMessage(handleMessage)
	relayproxy.WorkerPost(`["READY"]`)
}

// storeQuery sends S_QUERY to the Store Worker and calls fn with the eventsJSON response.
func storeQuery(filterRaw string, fn func(string)) {
	ps.nextStoreID++
	reqID := ps.nextStoreID
	ps.storeCBs[reqID] = fn
	relayproxy.WorkerPost(`["S_QUERY",` | helpers.Itoa(int64(reqID)) | `,` | filterRaw | `]`)
	rid := reqID
	relayproxy.WorkerSetTimeout(30000, func() {
		if _, ok2 := ps.storeCBs[rid]; ok2 {
			delete(ps.storeCBs, rid)
		}
	})
}

func handleMessage(msg string) {
	w := mw.New(msg)
	msgType := w.Str()
	switch msgType {
	case "REQ":
		subID := dup(w.Str())
		filterRaw := dup(w.Raw())
		handleReq(subID, filterRaw)
	case "CLOSE":
		subID := dup(w.Str())
		handleClose(subID)
	case "PROXY":
		subID := dup(w.Str())
		filterRaw := dup(w.Raw())
		relayURLs := dupStrs(w.Strs())
		handleProxy(subID, filterRaw, relayURLs, false)
	case "PROXY_LIVE":
		subID := dup(w.Str())
		filterRaw := dup(w.Raw())
		relayURLs := dupStrs(w.Strs())
		handleProxy(subID, filterRaw, relayURLs, true)
	case "EVENT":
		eventRaw := w.Raw()
		handleEventPublish(eventRaw)
	case "PUBLISH_TO":
		eventRaw := w.Raw()
		relayURLs := w.Strs()
		handlePublishTo(eventRaw, relayURLs)
	case "SET_WRITE_RELAYS":
		urls := dupStrs(w.Strs())
		handleSetWriteRelays(urls)
	case "SET_PUBKEY":
		ps.myPubkey = dup(w.Str())
	case "ENC_KEY":
		hexKey := w.Str()
		relayproxy.WorkerPost(`["S_ENC_KEY",` | jstr(hexKey) | `]`)
	case "SET_PROXY_EOSE_MS":
		ms := int32(w.Num())
		if ms > 0 {
			ps.eoseTimeoutMs = ms
		}
	case "MLS_SUB":
		urls := dupStrs(w.Strs())
		groupIDs := dupStrs(w.Strs())
		handleMLSSub(urls, groupIDs)
	case "MLS_UPDATE_GROUPS":
		groupIDs := dupStrs(w.Strs())
		handleMLSUpdateGroups(groupIDs)
	case "MLS_FETCH_KP":
		peer := dup(w.Str())
		urls := dupStrs(w.Strs())
		handleMLSFetchKP(peer, urls)
	// Store Worker response for storeQuery.
	case "SR_QUERY":
		reqID := int32(w.Num())
		eventsJSON := w.Raw()
		if fn, ok2 := ps.storeCBs[reqID]; ok2 {
			fn(eventsJSON)
			delete(ps.storeCBs, reqID)
		}
	}
}

func handleMLSSub(relayURLs, groupIDs []string) {
	println("[relay-proxy] handleMLSSub: urls=" | helpers.Itoa(int64(len(relayURLs))) | " groupIDs=" | helpers.Itoa(int64(len(groupIDs))) | " pubkey=" | ps.myPubkey[:16] | "...")
	if ps.myPubkey == "" || len(relayURLs) == 0 {
		println("[relay-proxy] handleMLSSub: no pubkey or no URLs, aborting")
		return
	}
	for rSubID := range ps.mlsSubIDs {
		for _, c := range ps.rpool.AllConns() {
			c.CloseSubscription(rSubID)
		}
	}
	ps.mlsSubIDs = map[string]bool{}
	ps.mlsGroupIDs = groupIDs

	for _, url := range relayURLs {
		url = normalizeRelayURL(url)
		if !isAllowedRelay(url) {
			println("[relay-proxy] handleMLSSub: blocked url=" | url)
			continue
		}
		suffix := urlSuffix(url)
		idP := "mlsp_" | suffix
		ps.mlsSubIDs[idP] = true
		c := getConn(url)
		println("[relay-proxy] handleMLSSub: sub " | idP | " kinds=[443,1059] #p=" | ps.myPubkey[:16] | "... on " | url)
		c.Subscribe(idP, []*nostr.Filter{{
			Kinds: []uint32{443, 1059},
			Tags:  map[string][]string{"#p": {ps.myPubkey}},
		}})
		if len(ps.mlsGroupIDs) > 0 {
			idH := "mlsh_" | suffix
			ps.mlsSubIDs[idH] = true
			println("[relay-proxy] handleMLSSub: sub " | idH | " kind=445 #h groups=" | helpers.Itoa(int64(len(ps.mlsGroupIDs))))
			c.Subscribe(idH, []*nostr.Filter{{
				Kinds: []uint32{445},
				Tags:  map[string][]string{"#h": ps.mlsGroupIDs},
			}})
		}
	}
}

func handleMLSUpdateGroups(groupIDs []string) {
	ps.mlsGroupIDs = groupIDs
	for _, c := range ps.rpool.AllConns() {
		suffix := urlSuffix(c.URL)
		idH := "mlsh_" | suffix
		c.CloseSubscription(idH)
		delete(ps.mlsSubIDs, idH)
		if len(ps.mlsGroupIDs) > 0 && ps.mlsSubIDs["mlsp_"|suffix] {
			ps.mlsSubIDs[idH] = true
			c.Subscribe(idH, []*nostr.Filter{{
				Kinds: []uint32{445},
				Tags:  map[string][]string{"#h": ps.mlsGroupIDs},
			}})
		}
	}
}

func handleMLSFetchKP(peer string, relayURLs []string) {
	println("[relay-proxy] handleMLSFetchKP: peer=" | peer[:16] | "... urls=" | helpers.Itoa(int64(len(relayURLs))))
	if peer == "" {
		println("[relay-proxy] handleMLSFetchKP: no peer")
		relayproxy.WorkerPost(`["MLS_KP_RESULT",` | jstr(peer) | `,null]`)
		return
	}
	// Resolve shared relays: fetch peer's kind 10002, intersect with ours.
	// Falls back to provided URLs if no intersection found.
	fallback := relayURLs
	if len(fallback) == 0 {
		fallback = ps.writeRelays
	}
	resolveSharedRelays(peer, fallback, func(urls []string) {
		if len(urls) == 0 {
			println("[relay-proxy] handleMLSFetchKP: no relays after resolution")
			relayproxy.WorkerPost(`["MLS_KP_RESULT",` | jstr(peer) | `,null]`)
			return
		}
		doKPFetch(peer, urls)
	})
}

func doKPFetch(peer string, relayURLs []string) {
	if ps.mlsKPFetchID != "" {
		closeKPFetchSubs()
	}
	ps.mlsKPFetchID = "mlskp_" | peer[:8]
	ps.mlsKPFetchPeer = peer
	ps.mlsKPFetchSubs = nil
	println("[relay-proxy] doKPFetch: peer=" | peer[:16] | "... urls=" | helpers.Itoa(int64(len(relayURLs))))
	for _, rawURL := range relayURLs {
		url := normalizeRelayURL(rawURL)
		if !isAllowedRelay(url) {
			println("[relay-proxy] doKPFetch: blocked url=" | url)
			continue
		}
		doKPFetchDirect(peer, url)
		return
	}
	println("[relay-proxy] doKPFetch: no valid URLs")
	ps.mlsKPFetchID = ""
	relayproxy.WorkerPost(`["MLS_KP_RESULT",` | jstr(peer) | `,null]`)
}

// doKPFetchDirect opens a dedicated WebSocket to fetch a single KP event.
// Bypasses the pool connection to avoid state interference.
func doKPFetchDirect(peer, url string) {
	subID := "kpq"
	reqMsg := `["REQ","` | subID | `",{"kinds":[443],"authors":["` | peer | `"],"limit":1}]`
	println("[relay-proxy] doKPFetchDirect: opening dedicated ws to " | url)
	println("[relay-proxy] doKPFetchDirect: REQ=" | reqMsg)

	done := false
	fetchPeer := peer

	wsConn := ws.Dial(url,
		func(connID int32, data string) {
			if done {
				return
			}
			println("[relay-proxy] doKPFetchDirect: recv len=" | helpers.Itoa(int64(len(data))))
			label, rSubID, payload := nostr.ParseRelayMessage(data)
			println("[relay-proxy] doKPFetchDirect: label=" | label | " subID=" | rSubID)
			if label == "EVENT" && rSubID == subID {
				ev := nostr.ParseEvent(payload)
				if ev != nil && ev.Kind == 443 {
					done = true
					evJSON := ev.ToJSON()
					println("[relay-proxy] doKPFetchDirect: FOUND KP for " | fetchPeer[:16] | "...")
					ps.mlsKPFetchID = ""
					ws.Close(ws.Conn(connID))
					relayproxy.WorkerPost(`["MLS_KP_RESULT",` | jstr(fetchPeer) | `,` | evJSON | `]`)
				}
			}
			if label == "EOSE" && rSubID == subID && !done {
				done = true
				println("[relay-proxy] doKPFetchDirect: EOSE with no KP for " | fetchPeer[:16] | "...")
				ps.mlsKPFetchID = ""
				ws.Close(ws.Conn(connID))
				relayproxy.WorkerPost(`["MLS_KP_RESULT",` | jstr(fetchPeer) | `,null]`)
			}
		},
		func(connID int32) {
			println("[relay-proxy] doKPFetchDirect: connected, sending REQ")
			ws.Send(ws.Conn(connID), reqMsg)
		},
		func(connID int32, code int32, reason string) {
			if !done {
				done = true
				println("[relay-proxy] doKPFetchDirect: ws closed code=" | helpers.Itoa(int64(code)))
				ps.mlsKPFetchID = ""
				relayproxy.WorkerPost(`["MLS_KP_RESULT",` | jstr(fetchPeer) | `,null]`)
			}
		},
		func(connID int32) {
			if !done {
				done = true
				println("[relay-proxy] doKPFetchDirect: ws error")
				ps.mlsKPFetchID = ""
				relayproxy.WorkerPost(`["MLS_KP_RESULT",` | jstr(fetchPeer) | `,null]`)
			}
		},
	)
	_ = wsConn

	relayproxy.WorkerSetTimeout(10000, func() {
		if !done {
			done = true
			println("[relay-proxy] doKPFetchDirect: TIMEOUT for " | fetchPeer[:16] | "...")
			ps.mlsKPFetchID = ""
			relayproxy.WorkerPost(`["MLS_KP_RESULT",` | jstr(fetchPeer) | `,null]`)
		}
	})
}

func closeKPFetchSubs() {
	for _, sid := range ps.mlsKPFetchSubs {
		for _, c := range ps.rpool.AllConns() {
			c.CloseSubscription(sid)
		}
	}
	ps.mlsKPFetchSubs = nil
}

// parseRelayList extracts relay URLs from a kind 10002 event.
// Returns URLs where the peer reads (or has no marker = both).
func parseRelayList(ev *nostr.Event) (ss []string) {
	var urls []string
	for _, tag := range ev.Tags {
		if len(tag) < 2 || tag[0] != "r" {
			continue
		}
		url := tag[1]
		if len(url) < 6 {
			continue
		}
		// If marker present, include only "read" or no marker (= both).
		if len(tag) >= 3 && tag[2] == "write" {
			continue
		}
		urls = mxutil.Ensure(urls, 1)
		urls = push(urls, normalizeRelayURL(url))
	}
	return urls
}

// resolveSharedRelays checks the cache for the peer's relay list.
// If cached, returns intersection immediately via cb.
// If not cached, fetches kind 10002 from connected relays, then calls cb.
func resolveSharedRelays(peer string, fallback []string, cb func([]string)) {
	if info, ok2 := ps.peerRelayCache[peer]; ok2 {
		shared := intersectRelays(ps.writeRelays, info.urls)
		println("[relay-proxy] resolveSharedRelays: cached peer=" | peer[:16] | "... shared=" | helpers.Itoa(int64(len(shared))))
		if len(shared) > 0 {
			cb(shared)
			return
		}
		cb(fallback)
		return
	}
	fetchPeerRelayList(peer, func(urls []string) {
		if len(urls) > 0 {
			ps.peerRelayCache[peer] = &peerRelayInfo{urls: urls}
			shared := intersectRelays(ps.writeRelays, urls)
			println("[relay-proxy] resolveSharedRelays: fetched peer=" | peer[:16] | "... peerURLs=" | helpers.Itoa(int64(len(urls))) | " shared=" | helpers.Itoa(int64(len(shared))))
			if len(shared) > 0 {
				cb(shared)
				return
			}
		} else {
			println("[relay-proxy] resolveSharedRelays: no 10002 for peer=" | peer[:16] | "... using fallback")
		}
		cb(fallback)
	})
}

func fetchPeerRelayList(peer string, cb func([]string)) {
	if ps.rlFetchID != "" {
		closeRLFetchSubs()
	}
	ps.rlFetchID = "rl10k_" | peer[:8]
	ps.rlFetchPeer = peer
	ps.rlFetchSubs = nil
	ps.rlFetchCB = cb
	ps.rlFetchEOSE = 0

	connURLs := ps.rpool.URLs()
	if len(connURLs) == 0 {
		connURLs = ps.writeRelays
	}
	for _, url := range connURLs {
		url = normalizeRelayURL(url)
		if !isAllowedRelay(url) {
			continue
		}
		suffix := urlSuffix(url)
		subID := ps.rlFetchID | "_" | suffix
		ps.rlFetchSubs = mxutil.Ensure(ps.rlFetchSubs, 1)
		ps.rlFetchSubs = push(ps.rlFetchSubs, subID)
		c := getConn(url)
		println("[relay-proxy] fetchPeerRelayList: sub " | subID | " kind=10002 author=" | peer[:16] | "... on " | url)
		c.Subscribe(subID, []*nostr.Filter{{
			Kinds:   []uint32{10002},
			Authors: []string{peer},
			Limit:   1,
		}})
	}
	fetchPrefix := ps.rlFetchID
	relayproxy.WorkerSetTimeout(5000, func() {
		if ps.rlFetchID == fetchPrefix {
			println("[relay-proxy] fetchPeerRelayList: TIMEOUT for " | ps.rlFetchPeer[:16] | "...")
			fn := ps.rlFetchCB
			closeRLFetchSubs()
			ps.rlFetchID = ""
			ps.rlFetchCB = nil
			if fn != nil {
				fn(nil)
			}
		}
	})
}

func closeRLFetchSubs() {
	for _, sid := range ps.rlFetchSubs {
		for _, c := range ps.rpool.AllConns() {
			c.CloseSubscription(sid)
		}
	}
	ps.rlFetchSubs = nil
}

func intersectRelays(ours, theirs []string) (ss []string) {
	var out []string
	for _, u := range ours {
		for _, t := range theirs {
			if normalizeRelayURL(u) == normalizeRelayURL(t) {
				out = mxutil.Ensure(out, 1)
				out = push(out, u)
				break
			}
		}
	}
	return out
}

func isHex(s string) (ok bool) {
	for i := 0; i < len(s); i++ {
		c := s[i]
		if !((c >= '0' && c <= '9') || (c >= 'a' && c <= 'f')) {
			return false
		}
	}
	return true
}

func handleEventPublish(eventRaw string) {
	ev := nostr.ParseEvent(eventRaw)
	if ev == nil {
		return
	}
	relayproxy.WorkerPost(`["S_PUT_EVENT",` | eventRaw | `]`)
	for _, url := range ps.writeRelays {
		if isAllowedRelay(url) {
			getConn(url).Publish(ev)
		}
	}
	relayproxy.WorkerPost(`["OK",` | jstr(ev.ID) | `,true,""]`)
}

func handlePublishTo(eventRaw string, relayURLs []string) {
	ev := nostr.ParseEvent(eventRaw)
	if ev == nil {
		println("[relay-proxy] handlePublishTo: parse failed")
		return
	}
	println("[relay-proxy] handlePublishTo: kind=" | helpers.Itoa(int64(ev.Kind)) | " id=" | ev.ID[:16] | "... to " | helpers.Itoa(int64(len(relayURLs))) | " relays")
	relayproxy.WorkerPost(`["S_PUT_EVENT",` | eventRaw | `]`)
	for _, url := range relayURLs {
		url = normalizeRelayURL(url)
		if isAllowedRelay(url) {
			c := getConn(url)
			oStr := "closed"
			if c.IsOpen() {
				oStr = "open"
			}
			println("[relay-proxy] handlePublishTo: publishing to " | url | " conn=" | oStr)
			c.Publish(ev)
		}
	}
	relayproxy.WorkerPost(`["OK",` | jstr(ev.ID) | `,true,""]`)
}

func handleSetWriteRelays(urls []string) {
	ps.writeRelays = nil
	for _, u := range urls {
		nu := normalizeRelayURL(u)
		if isAllowedRelay(nu) {
			ps.writeRelays = mxutil.Ensure(ps.writeRelays, 1)
			ps.writeRelays = push(ps.writeRelays, nu)
		}
	}
	println("[relay-proxy] SET_WRITE_RELAYS: " | helpers.Itoa(int64(len(ps.writeRelays))) | " relays")
	for _, u := range ps.writeRelays {
		println("[relay-proxy]   relay: " | u)
	}
}

func handleReq(subID, filterRaw string) {
	// Re-subscribing on an id restarts it, exactly as handleProxy does.
	cleanupProxy(subID)
	// The subscription record and the closure that answers the store query both
	// outlive this frame; allocate them in the root arena the maps live in.
	prev := runtime.CurrentArena()
	runtime.SovereignSetArena(runtime.RootArena())
	f := nostr.ParseFilter(filterRaw)
	if f == nil {
		runtime.SovereignRestoreArena(prev)
		return
	}
	ps.clientSubs[subID] = &clientSub{filter: f, filterRaw: filterRaw}
	// A plain REQ is the same subscription as a PROXY one with no relays, so it
	// goes through the same registry and fan-out: the remote half is complete
	// from the start and the store query alone ends the initial load. Keeping
	// it in ps.proxySubs means one EOSE policy and one place CLOSE has to clean up.
	ps.proxySubs[subID] = &proxySub{remoteIDs: map[string]bool{}, remoteDone: true}
	sid := subID
	storeQuery(filterRaw, func(eventsJSON string) {
		events := nostr.ParseEventsJSON(eventsJSON)
		for _, ev := range events {
			relayproxy.WorkerPost(`["EVENT",` | jstr(sid) | `,` | ev.ToJSON() | `]`)
		}
		if info, ok := ps.proxySubs[sid]; ok {
			info.localDone = true
			proxyMaybeEOSE(sid)
		}
	})
	runtime.SovereignRestoreArena(prev)
}

func handleClose(subID string) {
	delete(ps.clientSubs, subID)
	cleanupProxy(subID)
}

func handleProxy(subID, filterRaw string, relayURLs []string, live bool) {
	cleanupProxy(subID)

	// Everything this handler stores in the worker's maps - the subscription
	// record, its filter, the ids and filter text it is keyed by, and the
	// closures it hands to the timer service - outlives this frame and is read
	// by later messages. Allocate the handler's state in the root arena the
	// maps themselves live in. One PROXY message is one subscription, so the
	// scratch that stays is bounded by the subscription count.
	prev := runtime.CurrentArena()
	runtime.SovereignSetArena(runtime.RootArena())

	f := nostr.ParseFilter(filterRaw)
	if f == nil {
		runtime.SovereignRestoreArena(prev)
		return
	}
	ps.clientSubs[subID] = &clientSub{filter: f, filterRaw: filterRaw}

	sid := subID
	hasSearchField := false
	for i := 0; i+8 <= len(filterRaw); i++ {
		if filterRaw[i:i+8] == "\"search\"" {
			hasSearchField = true
			break
		}
	}
	if hasSearchField {
		// No local query runs for a search filter, so that source is complete.
	} else {
		storeQuery(filterRaw, func(eventsJSON string) {
			events := nostr.ParseEventsJSON(eventsJSON)
			for _, ev := range events {
				relayproxy.WorkerPost(`["EVENT",` | jstr(sid) | `,` | ev.ToJSON() | `]`)
			}
			if info, ok := ps.proxySubs[sid]; ok {
				info.localDone = true
				proxyMaybeEOSE(sid)
			}
		})
	}

	remoteIDs := map[string]bool{}
	base := "p_" | subID | "_"

	seen := map[string]bool{}
	deduped := []string{:0:len(relayURLs)}
	for _, url := range relayURLs {
		url = normalizeRelayURL(url)
		if seen[url] {
			continue
		}
		seen[url] = true
		deduped = push(deduped, url)
	}

	ps.proxySubs[subID] = &proxySub{
		remoteIDs:  remoteIDs,
		relayCount: len(deduped),
		live:       live,
	}

	for _, url := range deduped {
		if !isAllowedRelay(url) {
			continue
		}
		suffix := urlSuffix(url)
		rSubID := base | suffix
		remoteIDs[rSubID] = true
		ps.remoteToProxy[rSubID] = subID
		c := getConn(url)
		c.Subscribe(rSubID, []*nostr.Filter{f})
	}

	proxyID := subID
	if hasSearchField {
		// Nothing to wait for locally.
		if info, ok := ps.proxySubs[subID]; ok {
			info.localDone = true
		}
	}
	ps.proxySubs[subID].timer = relayproxy.WorkerSetTimeout(ps.eoseTimeoutMs, func() {
		info, ok := ps.proxySubs[proxyID]
		if !ok || info.done {
			return
		}
		info.remoteDone = true
		proxyMaybeEOSE(proxyID)
		if !info.done {
			// The timer has fired but the local store query has not answered.
			// End the initial load anyway after a short grace so the consumer
			// is never left waiting on it; 30s is too long for a feed.
			relayproxy.WorkerSetTimeout(1000, func() {
				cur, ok3 := ps.proxySubs[proxyID]
				if !ok3 || cur.done {
					return
				}
				cur.localDone = true
				proxyMaybeEOSE(proxyID)
			})
		}
		if !info.done || info.live {
			return
		}
		// Linger: keep the remote subs open for ps.lingerMs after EOSE so
		// events that arrive late from slow relays still get forwarded. The
		// consumer already got its EOSE so it can proceed, but stragglers
		// arriving during the linger window still match findProxySub and
		// reach the app. Closing immediately on EOSE timer was the source of
		// inconsistent profile resolution.
		info.timer = relayproxy.WorkerSetTimeout(ps.lingerMs, func() {
			cur, ok3 := ps.proxySubs[proxyID]
			if !ok3 {
				return
			}
			for rSubID := range cur.remoteIDs {
				delete(ps.remoteToProxy, rSubID)
				for _, c := range ps.rpool.AllConns() {
					c.CloseSubscription(rSubID)
				}
			}
			delete(ps.proxySubs, proxyID)
			delete(ps.clientSubs, proxyID)
		})
	})
	runtime.SovereignRestoreArena(prev)
}

// proxyMaybeEOSE emits the consumer's EOSE once the initial load is complete
// from both sources. The linger block that follows keeps the remote
// subscriptions open for a while so late stragglers still reach the app.
func proxyMaybeEOSE(proxyID string) {
	info, ok := ps.proxySubs[proxyID]
	if !ok || info.done {
		return
	}
	if !info.localDone || !info.remoteDone {
		return
	}
	info.done = true
	if _, ok2 := ps.clientSubs[proxyID]; ok2 {
		relayproxy.WorkerPost(`["EOSE",` | jstr(proxyID) | `]`)
	}
}

func cleanupProxy(proxyID string) {
	info, ok := ps.proxySubs[proxyID]
	if !ok {
		return
	}
	relayproxy.WorkerClearTimeout(info.timer)

	if !info.done {
		if _, ok2 := ps.clientSubs[proxyID]; ok2 {
			relayproxy.WorkerPost(`["EOSE",` | jstr(proxyID) | `]`)
		}
	}

	for rSubID := range info.remoteIDs {
		delete(ps.remoteToProxy, rSubID)
		for _, c := range ps.rpool.AllConns() {
			c.CloseSubscription(rSubID)
		}
	}
	delete(ps.proxySubs, proxyID)
}

func getConn(url string) (p *relay.Conn) {
	c := ps.rpool.Connect(url)
	if c.ScheduleReconnect == nil {
		wireConn(c)
	}
	return c
}

func wireConn(c *relay.Conn) {
	url := c.URL
	c.ScheduleReconnect = func(delayMs int32, fn func()) {
		relayproxy.WorkerSetTimeout(delayMs, fn)
	}
	c.SetOnEvent(func(rSubID string, ev *nostr.Event) {
		if ps.mlsSubIDs[rSubID] {
			evJSON := ev.ToJSON()
			println("[relay-proxy] MLS_EVENT: sub=" | rSubID | " kind=" | helpers.Itoa(int64(ev.Kind)) | " from=" | ev.PubKey[:16] | "...")
			relayproxy.WorkerPost(`["MLS_EVENT",` | evJSON | `]`)
			return
		}
		if ps.mlsKPFetchID != "" && len(rSubID) >= len(ps.mlsKPFetchID) && rSubID[:len(ps.mlsKPFetchID)] == ps.mlsKPFetchID && ev.Kind == 443 {
			evJSON := ev.ToJSON()
			peer := ps.mlsKPFetchPeer
			println("[relay-proxy] MLS_KP_RESULT: found KP for " | peer[:16] | "... on sub=" | rSubID)
			closeKPFetchSubs()
			ps.mlsKPFetchID = ""
			relayproxy.WorkerPost(`["MLS_KP_RESULT",` | jstr(peer) | `,` | evJSON | `]`)
			return
		}
		if ps.rlFetchID != "" && len(rSubID) >= len(ps.rlFetchID) && rSubID[:len(ps.rlFetchID)] == ps.rlFetchID && ev.Kind == 10002 {
			urls := parseRelayList(ev)
			peer := ps.rlFetchPeer
			println("[relay-proxy] RL_RESULT: peer=" | peer[:16] | "... urls=" | helpers.Itoa(int64(len(urls))))
			fn := ps.rlFetchCB
			closeRLFetchSubs()
			ps.rlFetchID = ""
			ps.rlFetchCB = nil
			if fn != nil {
				fn(urls)
			}
			return
		}
		// Proxy event path: skip CheckSig here - the Verify Supervisor in
		// wasm-host.mjs intercepts ["EVENT",...] and verifies before forwarding
		// to the app worker. Unverified events never reach app business logic.
		// A relay frame that cannot be routed to a client subscription is
		// reported rather than dropped silently: this path losing frames was
		// invisible because the only symptom was an empty feed.
		proxyID, info := findProxySub(rSubID)
		if info == nil {
			relayproxy.WorkerPost(`["PROXY_DROP",` | jstr(rSubID) | `,"nomap"]`)
			return
		}
		cs, ok := ps.clientSubs[proxyID]
		if !ok {
			relayproxy.WorkerPost(`["PROXY_DROP",` | jstr(rSubID) | `,"nosub"]`)
			return
		}
		_ = cs
		// No client-side filter re-check: the relay delivered this event for a
		// subscription it holds the filter for, and re-applying a cached copy
		// here dropped stored events whenever that cached filter was stale
		// (the second page load in a session reused the client subID).
		evJSON := ev.ToJSON()
		relayproxy.WorkerPost(`["S_PUT_EVENT",` | evJSON | `]`)
		relayproxy.WorkerPost(`["EVENT",` | jstr(proxyID) | `,` | evJSON | `]`)
		relayproxy.WorkerPost(`["SEEN_ON",` | jstr(ev.ID) | `,` | jstr(url) | `]`)
	})
	c.SetOnEOSE(func(rSubID string) {
		if ps.mlsSubIDs[rSubID] || (ps.mlsKPFetchID != "" && len(rSubID) >= len(ps.mlsKPFetchID) && rSubID[:len(ps.mlsKPFetchID)] == ps.mlsKPFetchID) {
			println("[relay-proxy] MLS EOSE: sub=" | rSubID | " relay=" | url)
		}
		if ps.rlFetchID != "" && len(rSubID) >= len(ps.rlFetchID) && rSubID[:len(ps.rlFetchID)] == ps.rlFetchID {
			ps.rlFetchEOSE++
			println("[relay-proxy] RL EOSE: sub=" | rSubID | " relay=" | url | " eose=" | helpers.Itoa(int64(ps.rlFetchEOSE)) | "/" | helpers.Itoa(int64(len(ps.rlFetchSubs))))
			if ps.rlFetchEOSE >= len(ps.rlFetchSubs) {
				fn := ps.rlFetchCB
				closeRLFetchSubs()
				ps.rlFetchID = ""
				ps.rlFetchCB = nil
				if fn != nil {
					fn(nil)
				}
			}
		}
	})
	c.SetOnOK(func(eventID string, ok bool, msg string) {
		okStr := "true"
		if !ok {
			okStr = "false"
			println("[relay-proxy] OK REJECTED: id=" | eventID[:16] | "... ok=" | okStr | " msg=" | msg | " relay=" | url)
		}
		relayproxy.WorkerPost(`["OK",` | jstr(eventID) | `,` | okStr | `,` | jstr(msg) | `]`)
		if ok {
			relayproxy.WorkerPost(`["SEEN_ON",` | jstr(eventID) | `,` | jstr(url) | `]`)
		}
	})
}

func findProxySub(rSubID string) (s string, p *proxySub) {
	proxyID, ok := ps.remoteToProxy[rSubID]
	if !ok {
		return "", nil
	}
	info, ok := ps.proxySubs[proxyID]
	if !ok {
		delete(ps.remoteToProxy, rSubID)
		return "", nil
	}
	return proxyID, info
}

func normalizeRelayURL(url string) (s string) {
	for len(url) > 0 && url[len(url)-1] == '/' {
		url = url[:len(url)-1]
	}
	return url
}

func isAllowedRelay(url string) (ok bool) {
	if len(url) >= 6 && url[:6] == "wss://" {
		return true
	}
	if len(url) >= 16 && url[:16] == "ws://localhost:" {
		return true
	}
	if len(url) >= 15 && url[:15] == "ws://127.0.0.1:" {
		return true
	}
	return false
}

func urlSuffix(url string) (s string) {
	// Extract host from wss://host/ or wss://host
	start := 0
	for i := 0; i < len(url); i++ {
		if url[i] == '/' && i+1 < len(url) && url[i+1] == '/' {
			start = i + 2
			break
		}
	}
	end := len(url)
	for i := start; i < len(url); i++ {
		if url[i] == '/' || url[i] == ':' {
			end = i
			break
		}
	}
	host := url[start:end]
	out := []byte{:0:len(host)}
	for i := 0; i < len(host); i++ {
		c := host[i]
		if (c >= 'a' && c <= 'z') || (c >= '0' && c <= '9') {
			out = push(out, c)
		}
	}
	if len(out) > 12 {
		out = out[len(out)-12:]
	}
	return string(out)
}

func jstr(s string) (sv string) { return helpers.JsonString(s) }

func dup(s string) (sv string) {
	if len(s) == 0 {
		return ""
	}
	b := []byte{:len(s)}
	copy(b, s)
	return string(b)
}

func dupStrs(ss []string) (ss2 []string) {
	if len(ss) == 0 {
		return nil
	}
	out := []string{:0:len(ss)}
	for _, s := range ss {
		out = push(out, dup(s))
	}
	return out
}