// Package unread owns process-local unread serving state.
package unread

import (
	"fmt"
	"hash/fnv"
	"strings"
	"sync"
	"time"
)

const (
	defaultSnapshotFlushInterval  = 30 * time.Second
	defaultSnapshotFlushBatchSize = 1000
	defaultHydrateMaxConcurrency  = 2
	defaultHydrateBatchSize       = 100

	defaultRecoveryRetryInterval    = 5 * time.Second
	defaultEventQueueSize           = 10000
	defaultAccountingBatchSize      = 500
	defaultMaxSyncAccountingMembers = 2000
	defaultShardCount               = 64
)

func supportedScopeType(scopeType ScopeType) bool {
	switch scopeType {
	case ScopeTypeChannel, ScopeTypeDM, ScopeTypeThread, ScopeTypeTask, ScopeTypeServiceAccount:
		return true
	default:
		return false
	}
}

// ServingState describes whether an unread value can be served as exact state
// or only as a best-effort hint.
type ServingState string

const (
	ServingStateReady           ServingState = "ready"
	ServingStateSnapshotServing ServingState = "snapshot_serving"
	ServingStateHydrating       ServingState = "hydrating"
	ServingStateStale           ServingState = "stale"
	ServingStateUnavailable     ServingState = "unavailable"
)

// ScopeType identifies the unread accounting owner for a scope.
type ScopeType string

const (
	ScopeTypeChannel        ScopeType = "channel"
	ScopeTypeDM             ScopeType = "dm"
	ScopeTypeThread         ScopeType = "thread"
	ScopeTypeTask           ScopeType = "task"
	ScopeTypeServiceAccount ScopeType = "service_account"
)

// Config contains operator-facing runtime settings plus internal safety
// budgets. Operator-facing values are normally copied from internal/config.
type Config struct {
	SnapshotFlushInterval  time.Duration
	SnapshotFlushBatchSize int
	HydrateMaxConcurrency  int
	HydrateBatchSize       int

	RecoveryRetryInterval    time.Duration
	EventQueueSize           int
	AccountingBatchSize      int
	MaxSyncAccountingMembers int
	ShardCount               int
}

// DefaultConfig returns the V1 unread runtime defaults.
func DefaultConfig() Config {
	return Config{
		SnapshotFlushInterval:  defaultSnapshotFlushInterval,
		SnapshotFlushBatchSize: defaultSnapshotFlushBatchSize,
		HydrateMaxConcurrency:  defaultHydrateMaxConcurrency,
		HydrateBatchSize:       defaultHydrateBatchSize,

		RecoveryRetryInterval:    defaultRecoveryRetryInterval,
		EventQueueSize:           defaultEventQueueSize,
		AccountingBatchSize:      defaultAccountingBatchSize,
		MaxSyncAccountingMembers: defaultMaxSyncAccountingMembers,
		ShardCount:               defaultShardCount,
	}
}

// New creates an owner-local unread runtime.
func New(cfg Config) (*Runtime, error) {
	cfg = withRuntimeDefaults(cfg)
	if err := cfg.validate(); err != nil {
		return nil, err
	}

	runtime := &Runtime{
		config: cfg,
		shards: make([]*shard, cfg.ShardCount),
	}
	for i := range runtime.shards {
		runtime.shards[i] = newShard()
	}
	return runtime, nil
}

func withRuntimeDefaults(cfg Config) Config {
	defaults := DefaultConfig()
	if cfg.SnapshotFlushInterval == 0 {
		cfg.SnapshotFlushInterval = defaults.SnapshotFlushInterval
	}
	if cfg.SnapshotFlushBatchSize == 0 {
		cfg.SnapshotFlushBatchSize = defaults.SnapshotFlushBatchSize
	}
	if cfg.HydrateMaxConcurrency == 0 {
		cfg.HydrateMaxConcurrency = defaults.HydrateMaxConcurrency
	}
	if cfg.HydrateBatchSize == 0 {
		cfg.HydrateBatchSize = defaults.HydrateBatchSize
	}
	if cfg.RecoveryRetryInterval == 0 {
		cfg.RecoveryRetryInterval = defaults.RecoveryRetryInterval
	}
	if cfg.EventQueueSize == 0 {
		cfg.EventQueueSize = defaults.EventQueueSize
	}
	if cfg.AccountingBatchSize == 0 {
		cfg.AccountingBatchSize = defaults.AccountingBatchSize
	}
	if cfg.MaxSyncAccountingMembers == 0 {
		cfg.MaxSyncAccountingMembers = defaults.MaxSyncAccountingMembers
	}
	if cfg.ShardCount == 0 {
		cfg.ShardCount = defaults.ShardCount
	}
	return cfg
}

func (cfg Config) validate() error {
	if cfg.SnapshotFlushInterval <= 0 {
		return fmt.Errorf("snapshot flush interval must be positive")
	}
	if cfg.SnapshotFlushBatchSize <= 0 {
		return fmt.Errorf("snapshot flush batch size must be positive")
	}
	if cfg.HydrateMaxConcurrency <= 0 {
		return fmt.Errorf("hydrate max concurrency must be positive")
	}
	if cfg.HydrateBatchSize <= 0 {
		return fmt.Errorf("hydrate batch size must be positive")
	}
	if cfg.RecoveryRetryInterval <= 0 {
		return fmt.Errorf("recovery retry interval must be positive")
	}
	if cfg.EventQueueSize <= 0 {
		return fmt.Errorf("event queue size must be positive")
	}
	if cfg.AccountingBatchSize <= 0 {
		return fmt.Errorf("accounting batch size must be positive")
	}
	if cfg.MaxSyncAccountingMembers <= 0 {
		return fmt.Errorf("max sync accounting members must be positive")
	}
	if cfg.ShardCount <= 0 {
		return fmt.Errorf("shard count must be positive")
	}
	return nil
}

// Runtime stores unread state for workspaces owned by this process.
type Runtime struct {
	config Config
	shards []*shard
}

// Config returns the resolved runtime configuration.
func (r *Runtime) Config() Config {
	if r == nil {
		return Config{}
	}
	return r.config
}

// RosterMember is one write-side eligible member in a scope roster.
type RosterMember struct {
	MemberID            string
	UserID              string
	LastMessageReadSeq  int64
	LastActivityReadSeq int64
	EffectiveFromSeq    int64
	NotificationState   string
	Visibility          string
	Ineligible          bool
}

// MessageStored describes a committed human-visible scope message event.
type MessageStored struct {
	WorkspaceID             string
	ScopeID                 string
	ScopeType               ScopeType
	Seq                     int64
	AuthorMemberID          string
	NotifyPrincipalMemberID string
	// AllowAuthorUnread lets trusted automation notify its Human owner when the
	// write-time audience includes that owner; ordinary Human messages leave it false.
	AllowAuthorUnread bool
	MessageGeneration int64
	// RecipientMemberIDs is an optional immutable event audience. Task
	// messages always set it: task participants are not implicitly notified,
	// while Watch, assignee attention, and a direct mention are evaluated once
	// at write time and survive later subscription changes.
	RecipientMemberIDs []string
}

// NonMessageActivityStored describes a committed high-signal durable activity
// event.
type NonMessageActivityStored struct {
	WorkspaceID        string
	ScopeID            string
	ScopeType          ScopeType
	Seq                int64
	ActorMemberID      string
	ActivityGeneration int64
	// RecipientMemberIDs is the immutable write-time Human attention snapshot.
	// It lets Task activity penetrate Unwatch for an assignee/direct mention
	// without turning that one event into a persistent Watch.
	RecipientMemberIDs []string
}

// MessageRewound describes committed rewind removal facts for one workspace.
type MessageRewound struct {
	WorkspaceID      string
	Scopes           []RewoundScope
	RewindGeneration int64
}

// RewoundScope describes one scope after a committed suffix rewind.
type RewoundScope struct {
	ScopeID         string
	LatestSeq       int64
	RemovedMessages []RewoundMessage
}

// RewoundMessage is one removed human-visible message fact.
type RewoundMessage struct {
	ScopeID                 string
	Seq                     int64
	AuthorMemberID          string
	NotifyPrincipalMemberID string
	// AllowAuthorUnread mirrors MessageStored for automation rewind accounting.
	AllowAuthorUnread bool
	MessageType       string
	CreatedAt         time.Time
}

// ReadCursorAdvanced describes a committed message read cursor update.
type ReadCursorAdvanced struct {
	WorkspaceID                 string
	MemberID                    string
	ScopeID                     string
	LastReadSeq                 int64
	MembershipGeneration        int64
	ManualUnreadGeneration      int64
	ManualUnreadGenerationKnown bool
}

// ActivityReadCursorAdvanced describes a committed activity read cursor update.
type ActivityReadCursorAdvanced struct {
	WorkspaceID          string
	MemberID             string
	ScopeID              string
	LastReadSeq          int64
	MembershipGeneration int64
}

// ManualUnreadChanged describes a committed manual unread state update.
type ManualUnreadChanged struct {
	WorkspaceID            string
	MemberID               string
	ScopeID                string
	ManualUnread           bool
	ManualUnreadGeneration int64
}

// ScopeInvalidated describes a committed policy, membership, visibility, or
// rewind mutation that makes loaded scope unread state no longer exact.
type ScopeInvalidated struct {
	WorkspaceID      string
	ScopeID          string
	MemberIDs        []string
	Reason           string
	Frontier         int64
	ActivityFrontier int64
	// FrontiersKnown marks Frontier and ActivityFrontier as an exact pair from
	// one committed Task hide transaction, including the valid zero/zero pair.
	// The expand release uses both to clear attention visible at hide time, but
	// deliberately allows only a later message to reopen the Task. Activity
	// reopening is activated only by the post-drain contract release after a
	// second hidden-row backfill. Other invalidations retain the legacy
	// roster-change suppression semantic.
	FrontiersKnown bool
	// Permanent records an irreversible whole-scope invalidation. Once set, the
	// owner-local runtime rejects late hydration and mutation callbacks for the
	// scope. Scope IDs are never reused, so the tombstone lives for the runtime's
	// lifetime. MemberIDs is ignored for permanent invalidations.
	Permanent bool
}

// Snapshot is a runtime snapshot value used for dirty flush and lazy hydrate.
type Snapshot struct {
	WorkspaceID                string
	MemberID                   string
	ScopeID                    string
	ScopeType                  ScopeType
	MessageUnreadCount         int64
	ActivityUnreadCount        int64
	ManualUnread               bool
	LastMessageReadSeq         int64
	LastActivityReadSeq        int64
	EffectiveFromSeq           int64
	LatestMessageAccountedSeq  int64
	LatestActivityAccountedSeq int64
	NotificationState          string
	MembershipVisibility       string
	ScopeMessageGeneration     int64
	ScopeActivityGeneration    int64
	MembershipGeneration       int64
	NotificationGeneration     int64
	ManualUnreadGeneration     int64
	RewindGeneration           int64
	UpdatedAt                  time.Time
	ServingState               ServingState
}

// FrontierMetadata is the metadata-only durable state used to classify a
// hydrated snapshot without reading message or activity rows.
type FrontierMetadata struct {
	WorkspaceID             string
	MemberID                string
	ScopeID                 string
	ScopeType               ScopeType
	LatestMessageSeq        int64
	LatestActivitySeq       int64
	ScopeMessageGeneration  int64
	ScopeActivityGeneration int64
	LastMessageReadSeq      int64
	LastActivityReadSeq     int64
	EffectiveFromSeq        int64
	NotificationState       string
	MembershipVisibility    string
	ManualUnread            bool
	MembershipGeneration    int64
	NotificationGeneration  int64
	ManualUnreadGeneration  int64
	RewindGeneration        int64
}

// ScopeUnreadState is the memory-backed unread result for one member/scope.
type ScopeUnreadState struct {
	WorkspaceID         string
	MemberID            string
	ScopeID             string
	ScopeType           ScopeType
	MessageUnreadCount  int64
	ActivityUnreadCount int64
	ManualUnread        bool
	ServingState        ServingState
	// StateVersion is process-local and monotonically increases for this exact
	// member/scope entry. It orders live projections emitted by one owner and
	// is deliberately not persisted across process restarts.
	StateVersion uint64
}

// isTaskScope reports whether the state belongs to a task scope.
func (s ScopeUnreadState) isTaskScope() bool {
	return s.ScopeType == ScopeTypeTask || strings.HasPrefix(strings.TrimSpace(s.ScopeID), "tsk_")
}

// Count returns the numeric unread badge value. Manual unread is intentionally
// separate so compatibility count maps do not fabricate a message count.
func (s ScopeUnreadState) Count() int64 {
	count := s.MessageUnreadCount + s.ActivityUnreadCount
	if count < 0 {
		return 0
	}
	return count
}

// HasUnread reports whether the state should contribute to workspace red dots.
func (s ScopeUnreadState) HasUnread() bool {
	return s.Count() > 0 || s.ManualUnread
}

// UnreadDelta is the realtime-safe unread state produced after memory
// accounting changes one member/scope entry.
type UnreadDelta struct {
	WorkspaceID                      string
	UserID                           string
	MemberID                         string
	ScopeID                          string
	ScopeType                        ScopeType
	Count                            int64
	ManualUnread                     bool
	HasUnread                        bool
	WorkspaceHasUnread               bool
	WorkspaceUnreadConversationCount int64
	ServingState                     ServingState
	StateVersion                     uint64
}

// ApplyResult summarizes runtime work for one event.
type ApplyResult struct {
	AccountedMembers int
	Batches          int
	Pending          bool
	Ignored          bool
	Deltas           []UnreadDelta
}

// PendingFrontier is the coalesced durable frontier that still needs bounded
// recovery or chunked accounting.
type PendingFrontier struct {
	WorkspaceID        string
	ScopeID            string
	ScopeType          ScopeType
	MessageSeq         int64
	ActivitySeq        int64
	MessageGeneration  int64
	ActivityGeneration int64
}

type shard struct {
	mu         sync.Mutex
	entries    map[entryKey]*entry
	aggregates map[memberKey]int
	scopes     map[scopeKey]*scopeState
	dirty      map[entryKey]struct{}
	pending    map[scopeKey]PendingFrontier

	suppressedRoster map[entryKey]suppressedRosterMember

	pendingMessages   map[scopeKey][]MessageStored
	pendingActivities map[scopeKey][]NonMessageActivityStored
	pendingOverflow   map[scopeKey]struct{}

	permanentScopeTombstones map[scopeKey]struct{}
}

func newShard() *shard {
	return &shard{
		entries:    make(map[entryKey]*entry),
		aggregates: make(map[memberKey]int),
		scopes:     make(map[scopeKey]*scopeState),
		dirty:      make(map[entryKey]struct{}),
		pending:    make(map[scopeKey]PendingFrontier),

		suppressedRoster: make(map[entryKey]suppressedRosterMember),

		pendingMessages:   make(map[scopeKey][]MessageStored),
		pendingActivities: make(map[scopeKey][]NonMessageActivityStored),
		pendingOverflow:   make(map[scopeKey]struct{}),

		permanentScopeTombstones: make(map[scopeKey]struct{}),
	}
}

type entryKey struct {
	workspaceID string
	memberID    string
	scopeID     string
}

type memberKey struct {
	workspaceID string
	memberID    string
}

type scopeKey struct {
	workspaceID string
	scopeID     string
}

type scopeState struct {
	rosterReady bool
	members     map[string]RosterMember
	memberIDs   []string
	scopeType   ScopeType
	// rosterGeneration is the current roster publish target. Every
	// refresh/invalidate-for-refresh bumps it and keeps rosterReady false. An
	// in-flight page-wise warm captures the target at BeginScopeRosterWarm into
	// rosterWarmGeneration and may only write pages / publish ready while the two
	// still match; otherwise it is a stale hydrate that read pre-refresh rows and
	// must not publish an authoritative roster.
	rosterGeneration     uint64
	rosterWarmGeneration uint64

	latestMessageAccountedSeq  int64
	latestActivityAccountedSeq int64
	scopeMessageGeneration     int64
	scopeActivityGeneration    int64
	taskMessageAudiences       map[int64]map[string]struct{}
	taskActivityAudiences      map[int64]map[string]struct{}
}

type entry struct {
	state  ScopeUnreadState
	userID string

	// snapshotResolved distinguishes an entry reconstructed from a durable
	// snapshot (or an explicit durable miss) from a control-only entry created
	// by an early read/manual callback after owner restart. Task attention must
	// not account a new event on top of the latter's implicit zero baseline.
	snapshotResolved        bool
	manualUnreadMutationSet bool

	// Snapshot counts are an opaque hydrated baseline: without loading durable
	// event rows the runtime cannot split them at an arbitrary cursor. Runtime
	// events accounted after that baseline retain their exact eligible seqs so
	// a delayed read cursor only removes events at or before that cursor.
	messageUnreadBaselineCount  int64
	messageUnreadBaselineSeq    int64
	messageUnreadSeqs           []int64
	activityUnreadBaselineCount int64
	activityUnreadBaselineSeq   int64
	activityUnreadSeqs          []int64

	lastMessageReadSeq         int64
	lastActivityReadSeq        int64
	effectiveFromSeq           int64
	latestMessageAccountedSeq  int64
	latestActivityAccountedSeq int64
	notificationState          string
	membershipVisibility       string
	scopeMessageGeneration     int64
	scopeActivityGeneration    int64
	membershipGeneration       int64
	notificationGeneration     int64
	manualUnreadGeneration     int64
	rewindGeneration           int64
}

type suppressedRosterMember struct {
	lastMessageReadSeq           int64
	lastActivityReadSeq          int64
	effectiveFromSeq             int64
	effectiveFromFloor           int64
	suppressedMessageThroughSeq  int64
	suppressedActivityThroughSeq int64
	sequenceFrontiersKnown       bool
	hasRosterBaseline            bool
	notificationState            string
	visibility                   string
}

func (r *Runtime) shardForWorkspace(workspaceID string) *shard {
	if r == nil || len(r.shards) == 0 {
		return nil
	}
	hash := fnv.New32a()
	_, _ = hash.Write([]byte(workspaceID))
	return r.shards[int(hash.Sum32())%len(r.shards)]
}

func cleanID(value string) string {
	return strings.TrimSpace(value)
}

func cleanState(value string) string {
	return strings.ToLower(strings.TrimSpace(value))
}