package main

import (
	"context"
	"encoding/json"
	"log"
	"net/http"
	"time"

	"github.com/coder/websocket"
)

// maxBlobBytes caps a single ciphertext payload. Plenty for a chat line plus
// AES-GCM overhead; keeps a misbehaving client from flooding the DB.
const maxBlobBytes = 64 * 1024

// client is one connected browser. send is the per-client outbound queue; the
// hub never blocks on a slow client — it drops them instead. userID lets the hub
// route a thread's traffic only to that thread's members.
type client struct {
	userID string
	send   chan []byte
}

// outbound is one fan-out: data delivered to every connected client whose user
// is in members (nil members = everyone, unused now that all traffic is
// thread-scoped).
type outbound struct {
	members map[string]bool
	data    []byte
}

// Hub fans out ciphertext blobs to connected clients. It only ever handles
// opaque bytes — it cannot read message contents — and routes each blob to the
// thread's members only.
type Hub struct {
	register   chan *client
	unregister chan *client
	broadcast  chan outbound
	control    chan func() // runs fn on the hub goroutine (safe map access)
	clients    map[*client]bool
}

func newHub() *Hub {
	return &Hub{
		register:   make(chan *client),
		unregister: make(chan *client),
		broadcast:  make(chan outbound, 256),
		control:    make(chan func()),
		clients:    make(map[*client]bool),
	}
}

func (h *Hub) run() {
	for {
		select {
		case fn := <-h.control:
			fn()
		case c := <-h.register:
			h.clients[c] = true
		case c := <-h.unregister:
			if h.clients[c] {
				delete(h.clients, c)
				close(c.send)
			}
		case o := <-h.broadcast:
			for c := range h.clients {
				if o.members != nil && !o.members[c.userID] {
					continue // not a member of this thread
				}
				select {
				case c.send <- o.data:
				default:
					// Backed-up client: drop it. Its read loop will clean up.
					delete(h.clients, c)
					close(c.send)
				}
			}
		}
	}
}

// shutdown closes every connected client so their browsers reconnect — to the
// new process, after a zero-downtime deploy (see docs/instance-router.md). It
// runs the close on the hub goroutine so it can't race broadcast/register. The
// closed send ends each client's writer, which closes the socket; the client's
// read loop then unregisters (a no-op, already removed).
func (h *Hub) shutdown() {
	done := make(chan struct{})
	h.control <- func() {
		for c := range h.clients {
			delete(h.clients, c)
			close(c.send)
		}
		close(done)
	}
	<-done
}

// broadcastToThread fans data out to the members of threadID only.
func (s *Server) broadcastToThread(threadID string, data []byte) {
	members, err := s.db.memberUserIDs(threadID)
	if err != nil {
		log.Println("broadcast: members:", err)
		return
	}
	set := make(map[string]bool, len(members))
	for _, u := range members {
		set[u] = true
	}
	s.hub.broadcast <- outbound{members: set, data: data}
}

type wsIn struct {
	Type      string `json:"type"`
	ThreadID  string `json:"thread_id"`
	Blob      string `json:"blob"`
	MessageID int64  `json:"message_id"`
	Emoji     string `json:"emoji"`
	// BlobIDs are the attachment blob ids a deleted message referenced. The server
	// can't read them from the (encrypted) message body, so the client — which can
	// — reports them on delete, and the server GCs those blobs scoped to the thread
	// (a client can only drop its own thread's blobs). See the "delete" case.
	BlobIDs []string `json:"blob_ids,omitempty"`
	// Ephemeral asks for relay-only delivery: fan the blob out to the thread's
	// members without storing or pushing it (presence-style traffic — typing,
	// live cursors). The flag lives on the envelope, not the ciphertext, so the
	// server can honour it while staying blind to the content.
	Ephemeral bool `json:"ephemeral,omitempty"`
}

type wsOut struct {
	Type      string `json:"type"`
	ThreadID  string `json:"thread_id,omitempty"`
	ID        int64  `json:"id"`
	CreatedAt int64  `json:"created_at,omitempty"`
	Blob      string `json:"blob"`
	EditedAt  int64  `json:"edited_at,omitempty"`
	Ephemeral bool   `json:"ephemeral,omitempty"`
}

// editWindow is how long after sending a message its author may still edit it.
// Enforced server-side against the message's created_at (see editMessage).
const editWindow = 15 * time.Minute

// activeWindow: a device counts as "active" (open + focused) if it signalled within
// this window. New messages skip push to a person active on ANY of their devices, so
// a chat you're reading on one device doesn't also notify your others.
const activeWindow = 45 * time.Second

func (s *Server) markActive(token string) {
	now := time.Now().Unix()
	s.active.Store(token, now)
	// Roll the signal up to the owner so suppression is per-person, not per-device.
	if uid, ok := s.db.deviceTokenUser(token); ok && uid != "" {
		s.activeUsers.Store(uid, now)
		// Fresh activity may flip an idle-away member back online (presence.go);
		// a pinned-away one stays away — presenceChanged resolves that.
		s.presenceChanged(uid)
	}
}

func (s *Server) isActive(token string) bool {
	if v, ok := s.active.Load(token); ok {
		return time.Since(time.Unix(v.(int64), 0)) < activeWindow
	}
	return false
}

// isUserActive reports whether the user is reading on any of their devices right now.
func (s *Server) isUserActive(uid string) bool {
	if uid == "" {
		return false
	}
	if v, ok := s.activeUsers.Load(uid); ok {
		return time.Since(time.Unix(v.(int64), 0)) < activeWindow
	}
	return false
}

// wsTypingOut relays a typing indicator. Like everything else it carries only an
// opaque blob (the sender's name + expiry ride encrypted inside) — the server
// never stores it and never pushes it; it's a transient presence hint.
type wsTypingOut struct {
	Type     string `json:"type"`
	ThreadID string `json:"thread_id,omitempty"`
	Blob     string `json:"blob"`
}

// wsReadOut tells a user's other devices their read watermark for a thread
// advanced, so their unread dots clear live. Routed only to that user's own
// connections — never to other members; reading habits aren't shared.
type wsReadOut struct {
	Type       string `json:"type"`
	ThreadID   string `json:"thread_id"`
	LastReadAt int64  `json:"last_read_at"`
}

func (s *Server) handleWS(w http.ResponseWriter, r *http.Request) {
	token := r.URL.Query().Get("token")
	if !s.tokenOK(token) {
		http.Error(w, "invalid invite", http.StatusForbidden)
		return
	}

	conn, err := websocket.Accept(w, r, nil) // same-origin check is on by default
	if err != nil {
		return
	}
	conn.SetReadLimit(maxBlobBytes + 1024)

	// The hub routes a thread's traffic to its members, so each client carries the
	// user it belongs to (empty for legacy non-device tokens — they receive nothing
	// thread-scoped until migrated).
	userID, _ := s.db.deviceTokenUser(token)
	cl := &client{userID: userID, send: make(chan []byte, 32)}
	s.hub.register <- cl
	// Presence: connecting counts as activity (a device that never sends
	// "active" frames — native, today — still reads online, then idles out).
	s.presence.addConn(userID)
	s.markActive(token)

	ctx := r.Context()

	// Writer: drains this client's queue. Closing send (by the hub) ends it.
	go func() {
		defer conn.CloseNow()
		for b := range cl.send {
			wctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
			err := conn.Write(wctx, websocket.MessageText, b)
			cancel()
			if err != nil {
				return
			}
		}
	}()

	// Reader: accepts new ciphertext, stores it, and broadcasts to everyone.
	// However the socket dies — clean close, error, or the hub dropping a
	// backed-up client — this defer runs, so the presence count stays true.
	defer func() {
		s.hub.unregister <- cl
		conn.CloseNow()
		s.presence.dropConn(userID)
		s.presenceChanged(userID)
	}()
	for {
		_, data, err := conn.Read(ctx)
		if err != nil {
			return
		}
		var in wsIn
		if json.Unmarshal(data, &in) != nil {
			continue
		}
		if in.Type == "active" {
			s.markActive(token) // this device is open + focused; mute its siblings' push
			continue
		}
		if in.Type != "msg" && in.Type != "edit" && in.Type != "delete" && in.Type != "typing" {
			continue
		}
		// Re-check the credential so a revoked invite / removed device can't keep
		// posting or reacting.
		if !s.tokenOK(token) {
			return
		}
		// All traffic is thread-scoped: you must be a member of the thread you're
		// posting/reacting to. This is the server-side gate on the social graph.
		if in.ThreadID == "" || !s.db.threadMember(in.ThreadID, userID) {
			continue
		}
		// A closed DM (issue #578) accepts no new traffic — messages, edits,
		// and typing are all communication. Deletes still pass: taking your
		// own words back is always allowed.
		if in.Type != "delete" {
			if closed, _ := s.db.dmClosed(in.ThreadID); closed {
				continue
			}
		}

		switch in.Type {
		case "typing":
			// Transient presence hint: relay the opaque blob to the thread's other
			// members (the sender ignores its own echo) without storing or pushing
			// it. The receiver decrypts the name + expiry and shows it briefly.
			//
			// This is a dedicated relay-only frame, like "active"/"ping" — typing
			// isn't message-shaped (no id, created_at, reactions). If more
			// ephemeral kinds appear later, consider generalising to a plaintext
			// `ephemeral` flag on the frame (read by the server to skip storage —
			// it must live on the envelope, not the ciphertext, which stays E2E).
			if in.Blob == "" || len(in.Blob) > maxBlobBytes {
				continue
			}
			out, _ := json.Marshal(wsTypingOut{Type: "typing", ThreadID: in.ThreadID, Blob: in.Blob})
			s.broadcastToThread(in.ThreadID, out)

		case "msg":
			if in.Blob == "" || len(in.Blob) > maxBlobBytes {
				continue
			}
			// The content-blind message brake (issue #578): far above any
			// human pace; a flood just stops landing. Admins exempt.
			if s.db.userRole(userID) != "admin" && !s.brake().messages.allow(userID) {
				continue
			}
			// Ephemeral: relay to the thread's members and forget — no store, no
			// push. This generalises the dedicated "typing" frame below for
			// mini-app presence traffic (the client marks the frame; content
			// stays opaque). Delivered with no id, so clients can't confuse it
			// with a stored message.
			if in.Ephemeral {
				out, _ := json.Marshal(wsOut{Type: "msg", ThreadID: in.ThreadID, Blob: in.Blob, Ephemeral: true})
				s.broadcastToThread(in.ThreadID, out)
				continue
			}
			m, err := s.db.insertMessage(in.ThreadID, in.Blob, userID)
			if err != nil {
				log.Println("insert message:", err)
				continue
			}
			out, _ := json.Marshal(wsOut{Type: "msg", ThreadID: in.ThreadID, ID: m.ID, CreatedAt: m.CreatedAt, Blob: m.Blob})
			s.broadcastToThread(in.ThreadID, out)
			// Notify the thread's other members (not the sender), best-effort, async.
			go s.pushNewMessage(m.Blob, in.ThreadID, token, m.ID, m.CreatedAt)

		case "edit":
			if in.MessageID <= 0 || in.Blob == "" || len(in.Blob) > maxBlobBytes ||
				!s.db.messageInThread(in.MessageID, in.ThreadID) {
				continue
			}
			// Edits share the message bucket — each one broadcasts to the whole
			// thread, so edit-churn is just message spam wearing a hat. (Typing
			// and ephemeral relays stay unthrottled on purpose: never stored,
			// never pushed, and burning message tokens on typing beats would
			// starve real sends.)
			if s.db.userRole(userID) != "admin" && !s.brake().messages.allow(userID) {
				continue
			}
			m, ok, err := s.db.editMessage(in.MessageID, in.Blob, userID, editWindow)
			if err != nil {
				log.Println("edit message:", err)
				continue
			}
			if !ok {
				continue // wrong author, too old, or unknown id — silently ignore
			}
			out, _ := json.Marshal(wsOut{Type: "edit", ThreadID: in.ThreadID, ID: m.ID, Blob: m.Blob, EditedAt: m.EditedAt})
			s.broadcastToThread(in.ThreadID, out)

		case "delete":
			if in.MessageID <= 0 || !s.db.messageInThread(in.MessageID, in.ThreadID) {
				continue
			}
			ok, err := s.db.deleteMessage(in.MessageID, userID)
			if err != nil {
				log.Println("delete message:", err)
				continue
			}
			if !ok {
				continue // not the author, or unknown id — silently ignore
			}
			// Reclaim the message's attachment blobs (scoped to this thread, so a
			// client can't drop another thread's blobs). Best-effort — a failure
			// here shouldn't block the delete fan-out.
			if len(in.BlobIDs) > 0 {
				if err := s.db.deleteBlobs(in.ThreadID, in.BlobIDs); err != nil {
					log.Println("gc attachment blobs:", err)
				}
			}
			out, _ := json.Marshal(wsOut{Type: "delete", ThreadID: in.ThreadID, ID: in.MessageID})
			s.broadcastToThread(in.ThreadID, out)

		}
	}
}