package spmetrics

import (
	"sync"

	"github.com/prometheus/client_golang/prometheus"
	"github.com/prometheus/client_golang/prometheus/promauto"
)

var viewersByStreamer = map[string]int{}
var viewersByProtocol = map[string]int{}
var viewersLock sync.RWMutex

var Viewers = promauto.NewGaugeVec(prometheus.GaugeOpts{
	Name: "streamplace_viewers",
	Help: "number of current viewers per user",
}, []string{"streamer"})

var ViewersTotal = promauto.NewGaugeVec(prometheus.GaugeOpts{
	Name: "streamplace_viewers_total",
	Help: "total number of viewers",
}, []string{"protocol"})

var StreamSessions = promauto.NewGaugeVec(prometheus.GaugeOpts{
	Name: "streamplace_stream_sessions",
	Help: "number of open stream sessions per streamer",
}, []string{"streamer"})

var SendSegmentCalls = promauto.NewGauge(prometheus.GaugeOpts{
	Name: "streamplace_send_segment_calls",
	Help: "total number of send segment calls currently in flight",
})

var SwarmPutCalls = promauto.NewGaugeVec(prometheus.GaugeOpts{
	Name: "streamplace_swarm_put_calls",
	Help: "total number of swarm put calls currently in flight",
}, []string{"streamer"})

var TranscodeAttemptsTotal = promauto.NewCounter(prometheus.CounterOpts{
	Name: "streamplace_transcode_attempts_total",
	Help: "total number of transcode attempts",
})

var TranscodeSuccessesTotal = promauto.NewCounter(prometheus.CounterOpts{
	Name: "streamplace_transcode_successes_total",
	Help: "total number of transcode successes",
})

var TranscodeSkippedTotal = promauto.NewCounter(prometheus.CounterOpts{
	Name: "streamplace_transcode_skipped_total",
	Help: "segments not transcoded because the transcoder was too far behind (see livepeer.MaxWaiting)",
})

var TranscodeErrorsTotal = promauto.NewCounter(prometheus.CounterOpts{
	Name: "streamplace_transcode_errors_total",
	Help: "total number of transcode errors",
})

var IngestNonIDRKeyframesTotal = promauto.NewCounterVec(prometheus.CounterOpts{
	Name: "streamplace_ingest_non_idr_keyframes_total",
	Help: "frames the parser flagged as keyframes that hold no IDR slice (open-GOP I-frames), kept inside their GoP instead of starting a segment",
}, []string{"streamer"})

var TranscodeSkippedNoKeyframeTotal = promauto.NewCounter(prometheus.CounterOpts{
	Name: "streamplace_transcode_skipped_no_keyframe_total",
	Help: "segments not sent to the transcoder because their video does not start with an IDR and its parameter sets (a transcoder cannot decode them and drops the session)",
})

var TranscodeManifestRotationsTotal = promauto.NewCounter(prometheus.CounterOpts{
	Name: "streamplace_transcode_manifest_rotations_total",
	Help: "times a stream moved to a fresh gateway manifest after the gateway refused every recent push (see livepeer.RotateAfterFailures)",
})

var TranscodeDuration = promauto.NewHistogramVec(prometheus.HistogramOpts{
	Name:    "streamplace_transcode_duration_ms",
	Help:    "duration of transcode in ms",
	Buckets: []float64{0, 250, 500, 750, 1000, 1250, 1500, 2000, 2500, 3000, 3500, 4000, 4500, 5000, 10000},
}, []string{"streamer"})

var SigningDuration = promauto.NewHistogramVec(prometheus.HistogramOpts{
	Name:    "streamplace_signing_duration_ms",
	Help:    "duration of transcode in ms",
	Buckets: []float64{0, 250, 500, 750, 1000, 1250, 1500, 2000, 2500, 3000, 3500, 4000, 4500, 5000, 10000, 20000, 30000, 60000},
}, []string{"streamer"})

// SegmentDeliveryDuration measures wall time from when the muxer hands a
// fresh segment to the sign callback through the end of validation —
// i.e. the latency the user actually waits on before the segment is
// available downstream. Wraps per-segment signing + ValidateMP4.
var SegmentDeliveryDuration = promauto.NewHistogramVec(prometheus.HistogramOpts{
	Name:    "streamplace_segment_delivery_duration_ms",
	Help:    "duration of sign + validate in ms (the user-visible per-segment latency)",
	Buckets: []float64{0, 250, 500, 750, 1000, 1250, 1500, 2000, 2500, 3000, 3500, 4000, 4500, 5000, 10000, 20000, 30000, 60000},
}, []string{"streamer"})

var QueuedTranscodeDuration = promauto.NewGaugeVec(prometheus.GaugeOpts{
	Name: "streamplace_queued_transcode_duration_ms",
	Help: "duration of transcode in ms, including time spent waiting",
}, []string{"streamer"})

var Version = promauto.NewCounterVec(prometheus.CounterOpts{
	Name: "streamplace_version",
	Help: "version of streamplace",
}, []string{"version"})

var WebsocketsOpen = promauto.NewGauge(prometheus.GaugeOpts{
	Name: "streamplace_websockets_open",
	Help: "number of open playback websockets",
})

var ReplicationWebsocketsOpen = promauto.NewGauge(prometheus.GaugeOpts{
	Name: "streamplace_replication_websockets_open",
	Help: "number of open replication websockets",
})

// --- syndication (outbound) -------------------------------------------------
//
// The inbound side is ReplicationWebsocketsOpen (peers pulling from this
// node). These are this node pulling from origins: what it does with each
// broadcast origin it hears about, how many pulls it holds open, and how
// often a pull fails. A node that is supposed to syndicate but shows only
// not_syndicated or self outcomes, or connect errors, is misconfigured or
// can't reach the origin.

// ReplicationOutboundOpen is the number of origin websockets this node is
// currently pulling segments from, per streamer.
var ReplicationOutboundOpen = promauto.NewGaugeVec(prometheus.GaugeOpts{
	Name: "streamplace_replication_outbound_open",
	Help: "outbound syndication websockets this node holds open to origin nodes, per streamer",
}, []string{"streamer"})

// BroadcastOriginsTotal counts broadcast origin records handled, by what was
// done with each: connect (a pull was opened), not_syndicated (the streamer
// isn't in the syndicate role), self (the origin is this node),
// already_connected, no_url (the record carries no websocket URL).
var BroadcastOriginsTotal = promauto.NewCounterVec(prometheus.CounterOpts{
	Name: "streamplace_broadcast_origins_total",
	Help: "broadcast origin records handled, by outcome (connect|not_syndicated|self|already_connected|no_url)",
}, []string{"outcome"})

// ReplicationConnectErrorsTotal counts outbound syndication pulls that
// failed: the dial, or a read after it.
var ReplicationConnectErrorsTotal = promauto.NewCounter(prometheus.CounterOpts{
	Name: "streamplace_replication_connect_errors_total",
	Help: "outbound syndication websocket dials or reads that failed",
})

var SegmentSubscriptionsOpen = promauto.NewGaugeVec(prometheus.GaugeOpts{
	Name: "streamplace_segment_subscriptions_open",
	Help: "number of open new segment subscriptions",
}, []string{"streamer", "rendition"})

var LabelerFirehosesConnected = promauto.NewGaugeVec(prometheus.GaugeOpts{
	Name: "streamplace_labeler_firehoses_connected",
	Help: "number of currently connected labeler firehoses",
}, []string{"labeler"})

// FirehoseRelaysConnected is 1 while a relay's firehose is connected and 0
// while it is reconnecting, labeled by relay host and protocol ("websocket" or
// "moq"). With multi-relay support this shows at a glance how many of the
// configured relays are currently feeding us.
var FirehoseRelaysConnected = promauto.NewGaugeVec(prometheus.GaugeOpts{
	Name: "streamplace_firehose_relays_connected",
	Help: "1 if the relay's firehose is currently connected, else 0",
}, []string{"relay", "protocol"})

// FirehoseEventsReceivedTotal counts every event a relay delivered, before
// cross-relay dedup, labeled by relay, protocol ("websocket"/"moq"), and kind.
// Comparing rates across relays tells you whether they carry the same volume;
// received minus deduped is each relay's first-seen (unique) contribution.
var FirehoseEventsReceivedTotal = promauto.NewCounterVec(prometheus.CounterOpts{
	Name: "streamplace_firehose_events_received_total",
	Help: "firehose events delivered by each relay before dedup, by relay/protocol/kind",
}, []string{"relay", "protocol", "kind"})

// FirehoseEventsDedupedTotal counts events dropped because the same commit (or
// identity update) already arrived from another relay, labeled by the relay that
// delivered the duplicate, its protocol, and event kind. A relay whose deduped
// count tracks its received count is fully redundant (a healthy mirror); a relay
// with a high first-seen count (received minus deduped) is contributing events
// the others are not — i.e. the relays are diverging.
var FirehoseEventsDedupedTotal = promauto.NewCounterVec(prometheus.CounterOpts{
	Name: "streamplace_firehose_events_deduped_total",
	Help: "firehose events dropped as cross-relay duplicates, by relay/protocol/kind",
}, []string{"relay", "protocol", "kind"})

// FirehoseBadFramesTotal counts firehose frames that failed strict decoding
// and were skipped (the connection stays up; only the frame is dropped),
// labeled by relay and protocol. A nonzero rate means a relay is delivering
// bytes that violate the atproto event schema — either corruption in the
// relay's serving path or genuinely invalid upstream data; the accompanying
// warn log carries the decoded context (DID, seq) for investigation.
var FirehoseBadFramesTotal = promauto.NewCounterVec(prometheus.CounterOpts{
	Name: "streamplace_firehose_bad_frames_total",
	Help: "firehose frames skipped because they failed decoding, by relay/protocol",
}, []string{"relay", "protocol"})

// FirehoseRelayHighSeq is the highest upstream sequence number seen from each
// relay. A WebSocket bsky relay and a MoQ relay that bridges the same upstream
// both carry bsky's original seq, so the difference between two relays' high_seq
// is a direct, units-of-events measure of how far apart they are: a small,
// roughly-constant skew means they are in sync (just a latency offset); a
// growing gap means one relay is falling behind or stuck.
var FirehoseRelayHighSeq = promauto.NewGaugeVec(prometheus.GaugeOpts{
	Name: "streamplace_firehose_relay_high_seq",
	Help: "highest upstream sequence number seen from each relay (cross-relay diff = parity/lag)",
}, []string{"relay", "protocol"})

// --- isolated ingest workers ------------------------------------------------

// IngestWorkerStarts counts isolated ingest worker subprocesses spawned, by
// transport ("mkv-fd" = fd-4 fallback, "mkv" = detached, "whip" = detached WHIP).
var IngestWorkerStarts = promauto.NewCounterVec(prometheus.CounterOpts{
	Name: "streamplace_ingest_worker_starts_total",
	Help: "isolated ingest worker subprocesses spawned, by transport",
}, []string{"transport"})

// IngestWorkerExits counts isolated ingest worker exits by transport and outcome
// ("clean" | "crash"). A rising crash rate is the signal that a stream is
// repeatedly faulting — the contained fault the node now survives but which would
// otherwise be invisible. A worker left running across a main shutdown is not an
// exit and is not counted; "resumed" is a worker reattached after a restart.
var IngestWorkerExits = promauto.NewCounterVec(prometheus.CounterOpts{
	Name: "streamplace_ingest_worker_exits_total",
	Help: "isolated ingest worker exits, by transport and outcome (clean|crash)",
}, []string{"transport", "outcome"})

// --- VOD processing ---------------------------------------------------------

// VODProcessAttemptsTotal increments once per task dequeued for VOD
// processing, labeled by the input storage backend (file/s3). Pair with
// VODProcessSuccessesTotal / VODProcessErrorsTotal to compute success
// rate.
var VODProcessAttemptsTotal = promauto.NewCounterVec(prometheus.CounterOpts{
	Name: "streamplace_vod_process_attempts_total",
	Help: "total number of VOD processing attempts (one per dequeued task)",
}, []string{"backend"})

var VODProcessSuccessesTotal = promauto.NewCounterVec(prometheus.CounterOpts{
	Name: "streamplace_vod_process_successes_total",
	Help: "total number of VOD processing runs that produced a content-addressed object",
}, []string{"backend"})

// VODProcessErrorsTotal is labeled by the stage that failed so we can
// see whether the bulk of failures are codec issues, S3 issues, etc.
// Stage values: open_source, gstreamer_pipeline, muxl_drain,
// s3_complete, content_address_copy.
var VODProcessErrorsTotal = promauto.NewCounterVec(prometheus.CounterOpts{
	Name: "streamplace_vod_process_errors_total",
	Help: "total number of VOD processing failures, by stage",
}, []string{"stage"})

// VODProcessDurationMS is the end-to-end wall time of one ProcessVOD
// call, from task dequeue to final CopyObject. The wide buckets cover
// realistic upload sizes (small phone clips through multi-hour
// long-form videos).
var VODProcessDurationMS = promauto.NewHistogram(prometheus.HistogramOpts{
	Name:    "streamplace_vod_process_duration_ms",
	Help:    "end-to-end wall time of one VOD processing run",
	Buckets: []float64{500, 1000, 2500, 5000, 10000, 30000, 60000, 120000, 300000, 600000, 1800000, 3600000},
})

// VODInputBytes / VODOutputBytes are observed once per successful run.
// Comparing them gives a transcode ratio (output/input) — e.g. opus->aac
// usually produces a modestly larger AAC payload, while many users will
// have inputs with surplus container overhead that gets trimmed.
var VODInputBytes = promauto.NewHistogram(prometheus.HistogramOpts{
	Name:    "streamplace_vod_input_bytes",
	Help:    "size in bytes of the user upload being processed",
	Buckets: prometheus.ExponentialBuckets(1<<20, 4, 8), // 1 MB ... 16 GB
})

var VODOutputBytes = promauto.NewHistogram(prometheus.HistogramOpts{
	Name:    "streamplace_vod_output_bytes",
	Help:    "size in bytes of the content-addressed fMP4 produced",
	Buckets: prometheus.ExponentialBuckets(1<<20, 4, 8),
})

// S3 helpers — instrumented for both the VOD pipeline and any future
// reusers of pkg/s3.ReaderAt / pkg/s3.MultipartWriter.

var S3ReaderAtOpensTotal = promauto.NewCounter(prometheus.CounterOpts{
	Name: "streamplace_s3_readerat_opens_total",
	Help: "total number of ranged GetObject opens issued by pkg/s3.ReaderAt",
})

var S3ReaderAtReadsTotal = promauto.NewCounterVec(prometheus.CounterOpts{
	Name: "streamplace_s3_readerat_reads_total",
	Help: "total ReadAt calls, by whether they hit the open body (sequential) or forced a reopen",
}, []string{"kind"}) // "sequential" or "seek"

var S3MultipartPartsUploadedTotal = promauto.NewCounter(prometheus.CounterOpts{
	Name: "streamplace_s3_multipart_parts_uploaded_total",
	Help: "total number of parts uploaded across all multipart uploads",
})

var S3MultipartBytesUploadedTotal = promauto.NewCounter(prometheus.CounterOpts{
	Name: "streamplace_s3_multipart_bytes_uploaded_total",
	Help: "total bytes written via pkg/s3.MultipartWriter",
})

func ViewerInc(user string, protocol string) {
	go func() {
		viewersLock.Lock()
		defer viewersLock.Unlock()
		viewersByStreamer[user]++
		viewersByProtocol[protocol]++
		Viewers.WithLabelValues(user).Set(float64(viewersByStreamer[user]))
		ViewersTotal.WithLabelValues(protocol).Set(float64(viewersByProtocol[protocol]))
	}()
}

func ViewerDec(user string, protocol string) {
	go func() {
		viewersLock.Lock()
		defer viewersLock.Unlock()
		viewersByStreamer[user]--
		if viewersByStreamer[user] == 0 {
			Viewers.DeleteLabelValues(user)
		} else {
			Viewers.WithLabelValues(user).Set(float64(viewersByStreamer[user]))
		}
		viewersByProtocol[protocol]--
		// Per-protocol series stay at 0 rather than vanishing: a node with
		// no viewers should read 0, not "no data". (Per-streamer ones are
		// unbounded, so those are still dropped.)
		ViewersTotal.WithLabelValues(protocol).Set(float64(viewersByProtocol[protocol]))
	}()
}

func GetViewCount(user string) int {
	viewersLock.RLock()
	defer viewersLock.RUnlock()
	return viewersByStreamer[user]
}

// A labelled metric only appears on /metrics once some label combination has
// been touched, so an idle node shows nothing at all for it — no way to tell
// "zero" from "not scraping". Touching the known label values (and an empty
// sentinel for per-streamer ones, which Prometheus reads as the label being
// absent) makes every one of these report 0 from boot. Per-relay and
// per-labeler ones are touched where those are configured.
func init() {
	for _, v := range []*prometheus.GaugeVec{Viewers, StreamSessions, SwarmPutCalls, ReplicationOutboundOpen} {
		v.WithLabelValues("")
	}
	SegmentSubscriptionsOpen.WithLabelValues("", "")
	for _, protocol := range []string{"webrtc", "hls", "local"} {
		ViewersTotal.WithLabelValues(protocol)
	}
	for _, outcome := range []string{"connect", "not_syndicated", "self", "already_connected", "no_url"} {
		BroadcastOriginsTotal.WithLabelValues(outcome)
	}
	for _, transport := range []string{"mp4-fd", "mp4", "whip"} {
		IngestWorkerStarts.WithLabelValues(transport)
		for _, outcome := range []string{"clean", "crash"} {
			IngestWorkerExits.WithLabelValues(transport, outcome)
		}
	}
	for _, backend := range []string{"file", "s3"} {
		VODProcessAttemptsTotal.WithLabelValues(backend)
		VODProcessSuccessesTotal.WithLabelValues(backend)
	}
}

// TouchRelay makes a relay's per-relay series exist at 0 before its first
// connection, so a relay that never connects is visible as 0 rather than
// missing.
func TouchRelay(relay, protocol string) {
	FirehoseRelaysConnected.WithLabelValues(relay, protocol)
	for _, kind := range []string{"commit", "identity"} {
		FirehoseEventsReceivedTotal.WithLabelValues(relay, protocol, kind)
	}
}