package ipfs

import (
	"context"
	"errors"
	"fmt"
	"log/slog"
	"net/http"
	"net/netip"
	"path/filepath"
	"sync"
	"time"

	boxobitswap "github.com/ipfs/boxo/bitswap"
	bsnet "github.com/ipfs/boxo/bitswap/network/bsnet"
	"github.com/ipfs/go-cid"
	ds "github.com/ipfs/go-datastore"
	dsync "github.com/ipfs/go-datastore/sync"
	"github.com/libp2p/go-libp2p"
	dht "github.com/libp2p/go-libp2p-kad-dht"
	"github.com/libp2p/go-libp2p/core/crypto"
	"github.com/libp2p/go-libp2p/core/host"
	"github.com/libp2p/go-libp2p/core/network"
	"github.com/libp2p/go-libp2p/core/peer"
	rcmgr "github.com/libp2p/go-libp2p/p2p/host/resource-manager"
	quic "github.com/libp2p/go-libp2p/p2p/transport/quic"
	"github.com/libp2p/go-libp2p/p2p/transport/tcp"
	libp2pwebrtc "github.com/libp2p/go-libp2p/p2p/transport/webrtc"
	"github.com/libp2p/go-libp2p/p2p/transport/websocket"
	webtransport "github.com/libp2p/go-libp2p/p2p/transport/webtransport"
	"github.com/libp2p/go-libp2p/x/rate"
	"github.com/multiformats/go-multiaddr"

	"atfs.dev/internal/store"
)

const (
	identityFileName = "identity.key"
	unixfsDirName    = "unixfs"

	// DefaultPort is the libp2p swarm (TCP+QUIC) listen port every atfs
	// instance binds — the standard kubo/IPFS default. It's no longer
	// configurable (see CLAUDE.md's "Bare-minimum configuration" rule):
	// never needed changing in practice, and dropping the knob leaves the
	// did:web document as the only setting still bound at startup.
	DefaultPort = 4001

	// reprovideInterval re-announces content periodically because DHT
	// provider records expire (~24-48h) — 12h keeps comfortable margin.
	reprovideInterval = 12 * time.Hour
	// retryInterval replaces reprovideInterval as the next pass's delay
	// when every provide in a pass failed — most likely a still-cold DHT
	// routing table — so content doesn't stay unannounced for a full
	// reprovideInterval.
	retryInterval = 5 * time.Minute
	// provideTimeout bounds a single CID's provide call, so one slow or
	// unreachable DHT lookup can't stall the whole announce pass.
	provideTimeout = time.Minute
	// bootstrapConnectTimeout bounds a single bootstrap peer dial.
	bootstrapConnectTimeout = 30 * time.Second
	// routingTableWaitTimeout bounds how long the first provide pass
	// waits for the DHT routing table to gain any peers before proceeding
	// anyway — see waitForRoutingTablePeers.
	routingTableWaitTimeout = 2 * time.Minute
	// routingTableWaitPoll is how often the wait re-checks routing table
	// size.
	routingTableWaitPoll = 2 * time.Second
	// provideCoalesceDelay is how long a provide-loop kick (see kickProvide)
	// waits, absorbing any further kicks, before triggering a pass. This
	// gives a burst of reachability transitions — and the identify pushes
	// that follow an address change — time to settle so the pass sees the
	// final address set rather than a half-updated one.
	provideCoalesceDelay = 10 * time.Second
	// announceConcurrency bounds how many AnnounceRoot-triggered DHT
	// Provide walks run at once — see the announceSem field. provideAll's
	// own periodic pass needs no such bound: it already announces one CID
	// at a time in a single sequential loop. AnnounceRoot is different
	// precisely because its callers don't share that loop — a burst of
	// uploads or completed pins can call it in quick succession, and
	// ATFS-d01i showed what an unauthenticated caller able to trigger it
	// once per listed file, per request, turns that into: a small cap
	// here is what stops any future caller from reintroducing an
	// unbounded fan-out, the same role internal/pin's fetchSem plays for
	// fetch attempts.
	announceConcurrency = 4

	// wsAnnounceTemplate builds the /tls/ws multiaddr announced for a
	// serviceDid domain (see announceWebSocket) — port 443, since that's
	// the only port a browser's wss:// can reach through an ordinary
	// TLS-terminating ingress (Cloudflare Tunnel and the like).
	wsAnnounceTemplate = "/dns4/%s/tcp/443/tls/ws"

	// wsSubnetConnLimit and wsSubnetConnRPS raise go-libp2p's own
	// defaults for inbound connections that share one address (see
	// newResourceManager) now that a browser can dial in over /ws:
	// go-libp2p allows 8 concurrent connections and 0.2 new ones per
	// second (burst 16) per /32 out of the box, sized for a swarm where
	// every peer is its own IP address. Behind a tunnel every browser
	// shares the tunnel's own address instead — go-libp2p has no
	// X-Forwarded-For handling anywhere — so the unmodified defaults
	// would refuse the 9th simultaneous uploader outright. These numbers
	// give real headroom for that without removing the limit altogether;
	// whether they hold up against the real Cloudflare tunnel is
	// unverified (see ATFS-twv3's MEASURE todos).
	wsSubnetConnLimit = 64
	wsSubnetConnRPS   = 2.0
)

// Node is atfs's IPFS-network participant: a libp2p host serving Bitswap
// over atfs's blob store and its UnixFS index, participating in the Amino
// DHT, periodically announcing its content's fetchable roots, and — via
// Pin — fetching content the other way, from providers or the network into
// the store.
type Node struct {
	host  host.Host
	dht   *dht.IpfsDHT
	bs    *boxobitswap.Bitswap
	index *Indexer
	store *store.Store

	// provideKick signals selfHealAndProvide's provide loop (via
	// runProvideLoop) to run a pass soon rather than waiting out the full
	// reprovideInterval — see kickProvide and logReachability. Buffered to
	// capacity 1; kickProvide always sends non-blockingly, so a burst of
	// reachability transitions collapses to a single pending kick.
	provideKick chan struct{}

	// ctx is the node's own long-lived context — derived from Start's ctx,
	// cancelled by cancel/Close — kept separately from any caller's
	// request-scoped context. AnnounceRoot needs it precisely because its
	// callers don't have one of their own that outlives them: an HTTP
	// handler's context dies the moment it returns, which is before
	// AnnounceRoot's background announce has any chance to run.
	ctx context.Context

	// providers is the HTTP client every provider attempt uses (see
	// newProviderClient), including the guard that keeps a caller-supplied
	// origin from reaching an address inside the operator's network. A
	// field rather than a package-level client so this package's own tests
	// can reach an httptest origin — over TLS, and by redirecting the
	// connection rather than by relaxing the guard, which is the only way
	// to exercise the real fetch path without a rule of it going untested.
	// Nothing outside this package can replace it: no configuration, no
	// record and no request reaches this field.
	providers *http.Client

	// announce performs a single CID's DHT provide — n.provideOne in
	// production, wired once n exists (see start) — used by AnnounceRoot. A
	// field rather than a direct call so tests can observe or replace it
	// without waiting out a real, or even a failing-fast offline, DHT
	// lookup.
	announce func(ctx context.Context, root cid.Cid) bool

	// announceSem bounds how many AnnounceRoot calls are actually running
	// n.announce at once — see announceConcurrency. Acquired inside
	// AnnounceRoot's own goroutine, never by its caller, so the wait for a
	// slot can't turn AnnounceRoot itself into something that blocks; the
	// acquire races n.ctx.Done() so a call still queued for a slot when
	// Close cancels the node gives up on it instead of leaking past
	// shutdown, the same shape internal/pin's fetchSem uses against a
	// fetch attempt's own deadline.
	announceSem chan struct{}

	cancel context.CancelFunc
	wg     sync.WaitGroup

	// closeMu guards AnnounceRoot's wg.Add against a concurrent Close.
	// AnnounceRoot is called from arbitrary caller goroutines, unlike the
	// two loops start() itself registers with wg before anyone could call
	// Close — so without this, a wg.Add landing after Close's own wg.Wait
	// has already returned zero would leak that goroutine past shutdown
	// (sync.WaitGroup's own documented hazard: adds with a positive delta
	// must happen before a Wait call they're meant to be counted by). Close
	// holds closeMu across setting closed and its own cancel, so an
	// AnnounceRoot that loses the race sees closed and skips instead.
	closeMu sync.Mutex
	closed  bool

	closeOnce sync.Once
	closeErr  error
}

// WebSocketConfig configures the browser-dialable /ws address this node
// mounts on its libp2p host, if any — see ATFS-twv3 and CLAUDE.md's
// "Serving" section. libp2p's WebSocket transport cannot be mounted into
// an existing mux (newListener and its ServeHTTP are unexported, and the
// transport always owns its own http.Server), so it's the other way
// round here too: the transport owns Port, and Handler serves everything
// on it that isn't a WebSocket upgrade. The zero value disables the whole
// thing — no /ws listen address, no fallback HTTP surface — which is what
// ATFS_HTTP_PORT=0 now means (see daemon.go).
type WebSocketConfig struct {
	// Port is the port the WebSocket listener binds — the same port a
	// bare net.Listen used to open directly (config.HTTP.Port). 0
	// disables the /ws listener entirely; Handler and AnnounceDomain are
	// then ignored.
	Port int
	// Handler serves every request on Port that isn't a WebSocket
	// upgrade: atfs's whole ordinary HTTP surface (xrpc, gateway,
	// did:web, the identity line), already wrapped in atfs's own
	// middleware (see internal/serve.WrapHandler). Required whenever
	// Port != 0.
	Handler http.Handler
	// AnnounceDomain, when non-empty, is the serviceDid domain (see
	// record.ParseDIDWebDomain) this instance's /ws address is announced
	// under, as /dns4/<AnnounceDomain>/tcp/443/tls/ws — added alongside,
	// not instead of, the plain listen address: a TLS-terminating
	// ingress (a Cloudflare Tunnel, say) still reaches this instance over
	// plain HTTP even though the browser dialed wss://, since libp2p's
	// Noise handshake runs inside the WebSocket frame regardless of where
	// TLS is terminated. Bound at startup, exactly like the served
	// did:web document: a later serviceDid change needs a restart to move
	// this (see daemon.go's did:web advisory log).
	AnnounceDomain string
}

// Start brings up the full IPFS stack: a libp2p host (TCP+QUIC always, plus
// a browser-dialable WebSocket listener when ws.Port != 0 — see
// WebSocketConfig), the Amino DHT (auto mode — server whenever this
// host is publicly reachable, client otherwise; standard IPFS bootstrap
// peers), Bitswap serving over st and its UnixFS index, and periodic DHT
// provider announcements. It walks the store to index any blob still
// missing a UnixFS DAG (self-healing) before its first announce pass.
//
// priv is loaded separately via LoadIdentity — see internal/daemon.Run's
// boot order, which needs the peer ID before the rest of the stack starts.
// dataDir is used only for the UnixFS index, stored alongside the identity
// file LoadIdentity(dataDir) reads.
//
// The returned Node shuts down cleanly when ctx is cancelled, or
// explicitly via Node.Close.
func Start(ctx context.Context, dataDir string, priv crypto.PrivKey, port int, st *store.Store, ws WebSocketConfig) (*Node, error) {
	return start(ctx, dataDir, priv, port, st, true, ws)
}

// start is Start's shared implementation. network=false skips dialing the
// real IPFS bootstrap peers, DHT bootstrapping, and the provide loop, so
// tests can exercise the libp2p host and Bitswap serving stack entirely
// offline and in-process — connected directly to another test host —
// without ever touching live DHT/bootstrap infrastructure.
func start(ctx context.Context, dataDir string, priv crypto.PrivKey, port int, st *store.Store, network bool, ws WebSocketConfig) (*Node, error) {
	index, err := OpenIndexer(filepath.Join(dataDir, unixfsDirName), st)
	if err != nil {
		return nil, err
	}

	// relay is the AutoRelay peer source's late-bound link to the DHT (see
	// relaySource) — the DHT can't exist yet at this point since dht.New
	// needs the host we're about to build, but AutoRelay needs its peer
	// source at host-construction time.
	var relay relaySource

	rm, err := newResourceManager()
	if err != nil {
		return nil, fmt.Errorf("ipfs: building resource manager: %w", err)
	}

	h, err := libp2p.New(
		libp2p.Identity(priv),
		libp2p.ListenAddrStrings(listenAddrs(port, ws.Port)...),
		// Explicit rather than libp2p.DefaultTransports: the /ws entry
		// needs WithHTTPHandler wired in (see wsTransportOpts), and
		// specifying any Transport option at all disables libp2p's own
		// "fill in DefaultTransports when none is set" fallback — so
		// every transport DefaultTransports would have registered has to
		// be listed here too, or TCP/QUIC/WebTransport/WebRTC silently
		// stop working.
		libp2p.Transport(tcp.NewTCPTransport),
		libp2p.Transport(quic.NewTransport),
		libp2p.Transport(websocket.New, wsTransportOpts(ws)...),
		libp2p.Transport(webtransport.New),
		libp2p.Transport(libp2pwebrtc.New),
		libp2p.AddrsFactory(announceWebSocket(ws)),
		libp2p.ResourceManager(rm),
		libp2p.NATPortMap(),
		libp2p.EnableHolePunching(),
		// AutoNAT v1's client (reachability detection) runs unconditionally
		// on any host; EnableNATService additionally serves dial-backs for
		// other peers. EnableAutoNATv2 does the same for the newer, more
		// precise per-address protocol — both are trivial-cost good-citizen
		// additions, not required for our own reachability detection.
		libp2p.EnableNATService(),
		libp2p.EnableAutoNATv2(),
		// Feeds AutoRelay candidate peers from the DHT routing table once it
		// exists; harmlessly empty until relay.set is called below, and
		// permanently in offline/test mode.
		libp2p.EnableAutoRelayWithPeerSource(autoRelayPeerSource(relay.candidates)),
	)
	if err != nil {
		return nil, fmt.Errorf("ipfs: starting libp2p host: %w", err)
	}

	// dht.ModeAuto — kubo's default — runs as a DHT client until AutoNAT
	// reports this host publicly reachable, then switches to server (and
	// back, if reachability regresses); see the mode switch in
	// go-libp2p-kad-dht's handleLocalReachabilityChangedEvent. That
	// depends on real AutoNAT reachability events, which the offline test
	// path (network=false) never gets — a never-connected host settles in
	// (and stays in) client mode. Tests connect two hosts directly and
	// don't exercise DHT routing at all, but pin ModeServer explicitly
	// there anyway: it's the honest, unsurprising choice for a "real DHT,
	// just offline" test double, and keeps prior behavior unchanged.
	mode := dht.ModeAuto
	if !network {
		mode = dht.ModeServer
	}
	memDS := dsync.MutexWrap(ds.NewMapDatastore())
	dhtOpts := []dht.Option{dht.Mode(mode), dht.Datastore(memDS)}
	if network {
		dhtOpts = append(dhtOpts, dht.BootstrapPeers(dht.GetDefaultBootstrapPeerAddrInfos()...))
	}
	kadDHT, err := dht.New(h, dhtOpts...)
	if err != nil {
		_ = h.Close()
		return nil, fmt.Errorf("ipfs: starting dht: %w", err)
	}
	relay.set(kadDHT)

	nodeCtx, cancel := context.WithCancel(ctx)

	// bitswap.New builds a client as well as a server. The client is what
	// Node.Pin fetches through; it never writes to bstore (it only reads it
	// for duplicate accounting), so handing it the read-only serving view
	// is safe — see newFetcher.
	bstore := NewBlockstore(st, index)
	bsNet := bsnet.NewFromIpfsHost(h)
	bs := boxobitswap.New(nodeCtx, bsNet, kadDHT, bstore)

	n := &Node{host: h, dht: kadDHT, bs: bs, index: index, store: st, providers: newProviderClient(), provideKick: make(chan struct{}, 1), announceSem: make(chan struct{}, announceConcurrency), ctx: nodeCtx, cancel: cancel}
	n.announce = n.provideOne
	index.announce = n.AnnounceRoot

	// Belt-and-braces: nodeCtx already dies when ctx does (it's derived
	// from ctx), which stops the background loop below, but Close also
	// needs to run to actually shut the host/DHT/bitswap down. This
	// watcher isn't tracked in n.wg — Close waits on n.wg, and Close is
	// what this goroutine calls, so it must stay outside that wait.
	go func() {
		<-ctx.Done()
		_ = n.Close()
	}()

	if network {
		n.connectBootstrapPeers(nodeCtx)
	}

	n.wg.Add(1)
	go func() {
		defer n.wg.Done()
		n.selfHealAndProvide(nodeCtx, network)
	}()

	n.wg.Add(1)
	go func() {
		defer n.wg.Done()
		n.logReachability(nodeCtx)
	}()

	return n, nil
}

// listenAddrs returns the swarm's TCP+QUIC listen addresses on port, plus
// a plain /ws listen address on wsPort when it's non-zero (see
// WebSocketConfig) — a browser can't speak TCP or QUIC, so /ws is the one
// address that makes this node dialable from one at all. TLS is not
// listened for directly: /ws is meant to sit behind a reverse proxy that
// terminates TLS upstream (a Cloudflare Tunnel, for atfs), and libp2p's own
// Noise handshake runs inside the WebSocket frame regardless — see
// announceWebSocket for the /tls/ws address actually announced to peers.
func listenAddrs(port, wsPort int) []string {
	addrs := []string{
		fmt.Sprintf("/ip4/0.0.0.0/tcp/%d", port),
		fmt.Sprintf("/ip6/::/tcp/%d", port),
		fmt.Sprintf("/ip4/0.0.0.0/udp/%d/quic-v1", port),
		fmt.Sprintf("/ip6/::/udp/%d/quic-v1", port),
	}
	if wsPort != 0 {
		addrs = append(addrs, fmt.Sprintf("/ip4/0.0.0.0/tcp/%d/ws", wsPort))
	}
	return addrs
}

// wsTransportOpts returns the websocket.Option list for libp2p.Transport
// to apply to the WebSocket transport — WithHTTPHandler wired to ws.Handler
// when the listener is enabled, or none at all when it's disabled (see
// WebSocketConfig), which leaves the transport registered but inert,
// exactly as it is today with no /ws listen address ever added.
//
// []any, not []websocket.Option: libp2p.Transport's variadic parameter is
// untyped (it validates each option against the constructor's real
// parameter type by reflection), so the options must already be boxed as
// any to spread into it.
func wsTransportOpts(ws WebSocketConfig) []any {
	if ws.Port == 0 {
		return nil
	}
	return []any{websocket.WithHTTPHandler(ws.Handler)}
}

// announceWebSocket builds the AddrsFactory that appends this instance's
// public /tls/ws address (see WebSocketConfig.AnnounceDomain) on top of
// whatever addresses the host would otherwise report — or leaves the
// address list untouched when there's no announce domain yet, or the /ws
// listener itself is disabled. AddrsFactory runs last, after libp2p's own
// reachability filtering (addrsManager.Addrs applies it after computing
// the dialable set), so this address is never pruned as unreachable the
// way a raw private/NATed address would be.
func announceWebSocket(ws WebSocketConfig) func([]multiaddr.Multiaddr) []multiaddr.Multiaddr {
	identity := func(addrs []multiaddr.Multiaddr) []multiaddr.Multiaddr { return addrs }
	if ws.Port == 0 || ws.AnnounceDomain == "" {
		return identity
	}
	announce, err := multiaddr.NewMultiaddr(fmt.Sprintf(wsAnnounceTemplate, ws.AnnounceDomain))
	if err != nil {
		// AnnounceDomain comes from record.ParseDIDWebDomain, already
		// validated as a bare hostname, so this would be a bug rather
		// than an operator mistake — log it and announce nothing extra
		// rather than fail startup over it.
		slog.Error("ipfs: building /ws announce address", "domain", ws.AnnounceDomain, "err", err)
		return identity
	}
	return func(addrs []multiaddr.Multiaddr) []multiaddr.Multiaddr {
		return append(addrs, announce)
	}
}

// newResourceManager builds the same resource manager go-libp2p would use
// by default (rcmgr.DefaultLimits, autoscaled, plus the bundled services'
// own limits — see libp2p.SetDefaultServiceLimits), except for a raised
// per-subnet inbound-connection ceiling and a matching connection-rate
// ceiling — see wsSubnetConnLimit/wsSubnetConnRPS.
func newResourceManager() (network.ResourceManager, error) {
	limits := rcmgr.DefaultLimits
	libp2p.SetDefaultServiceLimits(&limits)

	loopbackV4 := netip.MustParsePrefix("127.0.0.0/8")
	loopbackV6 := netip.MustParsePrefix("::1/128")

	return rcmgr.NewResourceManager(
		rcmgr.NewFixedLimiter(limits.AutoScale()),
		rcmgr.WithLimitPerSubnet(
			[]rcmgr.ConnLimitPerSubnet{{PrefixLength: 32, ConnCount: wsSubnetConnLimit}},
			[]rcmgr.ConnLimitPerSubnet{
				{PrefixLength: 56, ConnCount: wsSubnetConnLimit},
				{PrefixLength: 48, ConnCount: 8 * wsSubnetConnLimit},
			},
		),
		// WithConnRateLimiters replaces the whole rate limiter, unlike
		// WithLimitPerSubnet (which only overwrites the fields it's
		// given), so the loopback exemption go-libp2p's own default
		// carries has to be restated here too, or loopback traffic (every
		// test in this package included) would start getting rate-limited.
		rcmgr.WithConnRateLimiters(&rate.Limiter{
			NetworkPrefixLimits: []rate.PrefixLimit{
				{Prefix: loopbackV4, Limit: rate.Limit{}},
				{Prefix: loopbackV6, Limit: rate.Limit{}},
			},
			SubnetRateLimiter: rate.SubnetLimiter{
				IPv4SubnetLimits: []rate.SubnetLimit{
					{PrefixLength: 32, Limit: rate.Limit{RPS: wsSubnetConnRPS, Burst: wsSubnetConnLimit}},
				},
				IPv6SubnetLimits: []rate.SubnetLimit{
					{PrefixLength: 56, Limit: rate.Limit{RPS: wsSubnetConnRPS, Burst: wsSubnetConnLimit}},
					{PrefixLength: 48, Limit: rate.Limit{RPS: 4 * wsSubnetConnRPS, Burst: 8 * wsSubnetConnLimit}},
				},
				GracePeriod: time.Minute,
			},
		}),
	)
}

// connectBootstrapPeers fires off best-effort connects to the standard
// IPFS bootstrap peers so the DHT's routing table has somewhere to start.
// A peer that can't be reached (most commonly: this node is offline) is
// logged and otherwise ignored — startup must never depend on the
// network being reachable.
func (n *Node) connectBootstrapPeers(ctx context.Context) {
	for _, ai := range dht.GetDefaultBootstrapPeerAddrInfos() {
		go func(ai peer.AddrInfo) {
			cctx, cancel := context.WithTimeout(ctx, bootstrapConnectTimeout)
			defer cancel()
			if err := n.host.Connect(cctx, ai); err != nil {
				slog.Debug("ipfs: bootstrap connect failed", "peer", ai.ID, "err", err)
			}
		}(ai)
	}
}

// selfHealAndProvide indexes any blob still missing a complete UnixFS DAG
// (including healing a legacy manifest — see Indexer.isIndexed), finishes
// any blob deletion that crashed between unpin and delete, then (if
// network) bootstraps the DHT, waits for the routing table to gain peers,
// announces every blob's fetchable root, and re-announces on a ticker
// until ctx is cancelled — reprovideInterval normally, or the shorter
// retryInterval if a pass's every provide failed (see nextProvideInterval).
// It also runs an out-of-band pass promptly whenever logReachability
// signals a material address change via n.provideKick, so provider records
// pick up newly-dialable addresses well before the next scheduled pass.
//
// The index walk runs before the deletion sweep — see SweepDeletions —
// so the sweep's shared-chunk keep-set scan never reads a stale or legacy
// manifest.
func (n *Node) selfHealAndProvide(ctx context.Context, network bool) {
	if err := n.index.WalkAndIndex(ctx); err != nil {
		slog.Error("ipfs: startup index walk", "err", err)
	}
	if err := n.index.SweepDeletions(ctx); err != nil {
		slog.Error("ipfs: startup deletion sweep", "err", err)
	}
	if !network {
		return
	}

	if err := n.dht.Bootstrap(ctx); err != nil {
		slog.Warn("ipfs: dht bootstrap", "err", err)
	}

	// A provide pass fired the instant the DHT is bootstrapped routinely
	// finds an empty routing table — Provide then fails every CID with
	// "failed to find any peer in table" — so give the table a bounded
	// window to gain peers first. Proceeding after a timeout (rather than
	// waiting indefinitely) is deliberate: nextProvideInterval below still
	// catches an all-failed pass and retries soon.
	if !waitForRoutingTablePeers(ctx, n.dht.RoutingTable().Size, routingTableWaitPoll, routingTableWaitTimeout) && ctx.Err() == nil {
		slog.Info("ipfs: routing table still empty, proceeding with first provide pass anyway", "waited", routingTableWaitTimeout)
	}

	interval := nextProvideInterval(n.provideAll(ctx))
	runProvideLoop(ctx, interval, n.provideAll, n.provideKick, provideCoalesceDelay)
}

// runProvideLoop is selfHealAndProvide's post-bootstrap scheduling loop,
// factored out with injectable durations/channels so it's testable without
// a real DHT or real-length intervals. It runs provide (normally
// n.provideAll) on a ticker starting at initial and reset to
// nextProvideInterval(outcome) after every pass, plus an extra pass
// whenever kicks fires — coalesced via coalesceKicks so a burst of kicks in
// close succession triggers at most one extra pass. It returns when ctx is
// cancelled, including mid-coalesce.
func runProvideLoop(ctx context.Context, initial time.Duration, provide func(context.Context) provideOutcome, kicks <-chan struct{}, coalesceDelay time.Duration) {
	ticker := time.NewTicker(initial)
	defer ticker.Stop()
	for {
		select {
		case <-ctx.Done():
			return
		case <-ticker.C:
			ticker.Reset(nextProvideInterval(provide(ctx)))
		case <-kicks:
			if !coalesceKicks(ctx, kicks, coalesceDelay) {
				return
			}
			ticker.Reset(nextProvideInterval(provide(ctx)))
		}
	}
}

// coalesceKicks waits out delay after a provide-loop kick, absorbing any
// further kicks that arrive in that window into the same pending pass,
// before reporting true to trigger it. It reports false if ctx is
// cancelled first, so a kick that arrives during shutdown never blocks it.
func coalesceKicks(ctx context.Context, kicks <-chan struct{}, delay time.Duration) bool {
	timer := time.NewTimer(delay)
	defer timer.Stop()
	for {
		select {
		case <-ctx.Done():
			return false
		case <-timer.C:
			return true
		case <-kicks:
			// Absorbed into the same pending pass; keep waiting out delay.
		}
	}
}

// waitForRoutingTablePeers polls sizeFn (the DHT routing table's peer
// count, injected so this is testable without a real DHT) until it reports
// at least one peer, ctx is cancelled, or timeout elapses. It reports
// whether peers were found.
func waitForRoutingTablePeers(ctx context.Context, sizeFn func() int, poll, timeout time.Duration) bool {
	if sizeFn() > 0 {
		return true
	}

	deadline := time.NewTimer(timeout)
	defer deadline.Stop()
	ticker := time.NewTicker(poll)
	defer ticker.Stop()
	for {
		select {
		case <-ctx.Done():
			return false
		case <-deadline.C:
			return false
		case <-ticker.C:
			if sizeFn() > 0 {
				return true
			}
		}
	}
}

// provideOutcome summarizes one provideAll pass: how many roots it tried
// to announce and how many of those succeeded. selfHealAndProvide feeds it
// to nextProvideInterval to pick the next pass's delay.
type provideOutcome struct {
	attempted int
	succeeded int
}

// provideAll announces the fetchable root of every blob in the store: the
// blessed CID itself for blobs at or under MaxRawBlockSize, the UnixFS
// root for larger, indexed ones. A failure announcing any single CID
// (most commonly: this node is offline) is logged and skipped — it must
// never crash the node or abort the rest of the pass.
func (n *Node) provideAll(ctx context.Context) provideOutcome {
	var out provideOutcome
	for c, err := range n.store.List() {
		if ctx.Err() != nil {
			return out
		}
		if err != nil {
			slog.Error("ipfs: provide: listing store", "err", err)
			continue
		}

		root, ok, err := n.rootToProvide(c)
		if err != nil {
			slog.Error("ipfs: provide: resolving root", "cid", c, "err", err)
			continue
		}
		if !ok {
			continue
		}

		out.attempted++
		if n.provideOne(ctx, root) {
			out.succeeded++
		}
	}
	return out
}

// nextProvideInterval picks the delay before the next provide pass: the
// short retryInterval if the pass had at least one CID to announce and
// every one of them failed (most likely a DHT routing table that's still
// cold), so content doesn't sit unannounced for a full reprovideInterval;
// reprovideInterval otherwise, including when there was nothing to
// provide.
func nextProvideInterval(out provideOutcome) time.Duration {
	if out.attempted > 0 && out.succeeded == 0 {
		return retryInterval
	}
	return reprovideInterval
}

// rootToProvide returns the CID that should be announced for blob c: its
// UnixFS root if it's been indexed, or the blessed CID itself if it's
// small enough to need no DAG. It reports ok=false for a large blob that
// hasn't been indexed yet (self-heal hasn't caught up, or keeps failing)
// — there's nothing valid to announce for it yet.
func (n *Node) rootToProvide(c cid.Cid) (cid.Cid, bool, error) {
	if root, ok, err := n.index.RootFor(c); err != nil {
		return cid.Undef, false, err
	} else if ok {
		return root, true, nil
	}

	stat, err := n.store.Stat(c)
	if err != nil {
		return cid.Undef, false, err
	}
	if stat.Size > MaxRawBlockSize {
		return cid.Undef, false, nil
	}
	return c, true, nil
}

// provideOne announces a single root, bounded by provideTimeout, and
// reports whether it succeeded.
func (n *Node) provideOne(ctx context.Context, root cid.Cid) bool {
	pctx, cancel := context.WithTimeout(ctx, provideTimeout)
	defer cancel()
	if err := n.dht.Provide(pctx, root, true); err != nil {
		slog.Warn("ipfs: provide failed", "cid", root, "err", err)
		return false
	}
	return true
}

// AnnounceRoot triggers an immediate, best-effort DHT announcement of
// root — a blob's fetchable IPFS root, the same CID uploadFile reports as
// ipfsRoot and a completed pin ends up serving — right after it becomes
// available, rather than leaving it to wait for the next periodic provide
// pass (up to reprovideInterval away; direct-connect Bitswap still works
// in the meantime regardless, since it never depends on the DHT). See
// uploadfile.go's resolveIPFSRoot (via Indexer.AnnounceRoot) and this
// package's own Pin for its two callers — deliberately not xrpc's
// listFiles, which resolves a root for its response the same way but as a
// pure read (see xrpc's rootForListing): ATFS-d01i is what happened when a
// shared helper blurred that line and let a public, unauthenticated GET
// reach this method once per listed file, per request.
//
// It never blocks the caller and never fails visibly: the announce runs on
// its own tracked goroutine, through n.announce (n.provideOne in
// production — the same single-CID provide every periodic pass itself
// uses), so it's bounded by provideTimeout and any failure — most commonly
// a still-cold DHT, or this node being offline — is logged there and
// otherwise dropped. That goroutine waits for a slot in announceSem before
// calling n.announce (see announceConcurrency), so a burst of simultaneous
// callers still runs only a handful of concurrent DHT walks rather than one
// per call; the wait races n.ctx.Done(), so a call still queued for a slot
// when the node starts closing gives up on it rather than outliving
// shutdown. The periodic loop already covers the CID from here on: by the
// time this is called the blob is in the store, and provideAll walks the
// whole store on every pass.
//
// A no-op once the node is closing, so Close never waits on an
// announcement that only just started (see closeMu).
func (n *Node) AnnounceRoot(root cid.Cid) {
	n.closeMu.Lock()
	if n.closed {
		n.closeMu.Unlock()
		return
	}
	n.wg.Add(1)
	n.closeMu.Unlock()

	go func() {
		defer n.wg.Done()
		select {
		case n.announceSem <- struct{}{}:
			defer func() { <-n.announceSem }()
		case <-n.ctx.Done():
			return
		}
		n.announce(n.ctx, root)
	}()
}

// Index is this node's UnixFS index. It satisfies both xrpc.Indexer and
// gateway.Resolver, so the HTTP transport wiring (separate, later work)
// can pass it straight to xrpc.New and gateway.New.
func (n *Node) Index() *Indexer { return n.index }

// PeerID is this node's libp2p peer ID.
func (n *Node) PeerID() peer.ID { return n.host.ID() }

// Addrs is this node's current listen addresses.
func (n *Node) Addrs() []multiaddr.Multiaddr { return n.host.Addrs() }

// Close shuts the node down: the background index/provide loop, then
// Bitswap, the DHT, and finally the libp2p host. It's idempotent and safe
// to call multiple times, or concurrently with ctx cancellation (Start's
// own ctx-triggered shutdown calls this too).
func (n *Node) Close() error {
	n.closeOnce.Do(func() {
		n.closeMu.Lock()
		n.closed = true
		n.closeMu.Unlock()

		n.cancel()
		n.wg.Wait()

		var errs []error
		if err := n.bs.Close(); err != nil {
			errs = append(errs, err)
		}
		if err := n.dht.Close(); err != nil {
			errs = append(errs, err)
		}
		if err := n.host.Close(); err != nil {
			errs = append(errs, err)
		}
		n.closeErr = errors.Join(errs...)
	})
	return n.closeErr
}