Commilitia-Drop/internal/transfer/service.go

package transfer

import (
	"context"
	"crypto/rand"
	"encoding/hex"
	"errors"
	"fmt"
	"log/slog"
	"time"

	"commilitia.net/cdrop/internal/apns"
	"commilitia.net/cdrop/internal/db"
	"commilitia.net/cdrop/internal/hub"
	"commilitia.net/cdrop/internal/push"
)

var (
	ErrNotFound          = errors.New("transfer: session not found")
	ErrForbidden         = errors.New("transfer: not your session")
	ErrInvalidTransition = errors.New("transfer: invalid state transition")
	ErrRelayUnavailable  = errors.New("transfer: relay unavailable")
)

// SessionView is the session shape pushed to clients (omits internal columns).
type SessionView struct {
	ID         string `json:"id"`
	SenderName string `json:"sender_name"`
	FileName   string `json:"file_name"`
	FileSize   int64  `json:"file_size"`
	FileSHA256 string `json:"file_sha256,omitempty"`
}

// RelayController is the subset of *relay.Manager the transfer service drives:
// it reserves a relay slot when a transfer enters RELAY_ACTIVE (Register) and
// frees it when the transfer reaches a terminal state (Release). Declared as an
// interface so the transfer package doesn't gain a hard import of relay (avoids
// cycles in tests).
type RelayController interface {
	Register(id, userID, sender, receiver string) error
	Release(sessionID string)
}

type nopRelay struct{}

func (nopRelay) Register(_, _, _, _ string) error { return nil }
func (nopRelay) Release(string)                   {}

type Service struct {
	queries *db.Queries
	hub     *hub.Hub
	relay   RelayController
	push    *push.Sender // inert when Web Push is unconfigured; nil-safe
	apns    *apns.Sender // inert when APNs is unconfigured; nil-safe

	now   func() time.Time
	newID func() string
}

func NewService(queries *db.Queries, h *hub.Hub, r RelayController, pushSender *push.Sender, apnsSender *apns.Sender) *Service {
	if r == nil {
		r = nopRelay{}
	}
	return &Service{
		queries: queries,
		hub:     h,
		relay:   r,
		push:    pushSender,
		apns:    apnsSender,
		now:     time.Now,
		newID:   defaultNewID,
	}
}

func defaultNewID() string {
	var b [16]byte
	if _, err := rand.Read(b[:]); err != nil {
		// crypto/rand only fails if the OS RNG is broken; panicking is correct.
		panic(fmt.Sprintf("crypto/rand: %v", err))
	}
	return hex.EncodeToString(b[:])
}

// InitParams are the inputs to Init.
type InitParams struct {
	UserID       string
	SenderName   string
	ReceiverName string
	FileName     string
	FileSize     int64
	FileSHA256   string
}

// Init creates a new PENDING session and pushes transfer:incoming to the receiver.
// Returns the new session ID. Receiver offline does not fail Init — the receiver
// will see the session next time it connects via /api/devices or a fresh init.
// Init returns (sessionID, receiverOnline, error). receiverOnline is whether the receiver had a
// live SSE at init time — false means the receiver is offline and was sent a wake push, so the
// sender should wait for it to come online before offering (rather than fail immediately).
func (s *Service) Init(ctx context.Context, p InitParams) (string, bool, error) {
	if p.SenderName == "" || p.ReceiverName == "" || p.FileName == "" {
		return "", false, errors.New("init: sender_name, receiver_name, file_name required")
	}
	if p.SenderName == p.ReceiverName {
		return "", false, errors.New("init: sender and receiver must differ")
	}
	// brief §2 explicitly says "不限文件大小"; 0-byte is legal (empty file
	// transfer still needs control frames). Negatives are nonsense.
	if p.FileSize < 0 {
		return "", false, errors.New("init: file_size cannot be negative")
	}

	id := s.newID()
	createdAt := s.now().Unix()

	var sha *string
	if p.FileSHA256 != "" {
		v := p.FileSHA256
		sha = &v
	}

	if err := s.queries.CreateTransferSession(ctx, db.CreateTransferSessionParams{
		ID:           id,
		UserID:       p.UserID,
		SenderName:   p.SenderName,
		ReceiverName: p.ReceiverName,
		FileName:     p.FileName,
		FileSize:     p.FileSize,
		FileSha256:   sha,
		CreatedAt:    createdAt,
	}); err != nil {
		return "", false, fmt.Errorf("create session: %w", err)
	}

	view := SessionView{
		ID:         id,
		SenderName: p.SenderName,
		FileName:   p.FileName,
		FileSize:   p.FileSize,
		FileSHA256: p.FileSHA256,
	}
	autoAccept := p.FileSize <= AutoAcceptThreshold

	delivered := s.hub.SendTo(p.UserID, p.ReceiverName, hub.Event{
		Type: "transfer:incoming",
		Data: map[string]any{
			"session":     view,
			"auto_accept": autoAccept,
		},
	})
	// Receiver's page is closed → no SSE → raise a system notification instead.
	// Both Web Push (browser/PWA) and APNs (iOS) are attempted; each sender
	// only fires for its own platform rows, so firing both is always correct.
	if !delivered {
		n := push.Notification{
			Type:   push.KindTransferIncoming,
			Params: map[string]string{"sender": p.SenderName, "filename": p.FileName},
			Tag:    "transfer:" + id,
			URL:    "/",
		}
		if s.push.Enabled() {
			go s.push.Notify(context.Background(), p.UserID, p.ReceiverName, n)
		}
		if s.apns.Enabled() {
			go s.apns.Notify(context.Background(), p.UserID, p.ReceiverName, n)
		}
	}

	return id, delivered, nil
}

// pendingTransferWindowSec bounds how far back RedeliverPendingTo looks: a transfer older than
// this (the sender's wait window plus margin) the sender has long since given up on.
const pendingTransferWindowSec = 180

// RedeliverPendingTo re-sends transfer:incoming for the receiver's still-PENDING sessions when it
// (re)connects — e.g. an iOS device that was offline at Init and is now woken by the incoming
// push. Without this the woken receiver never arms its P2P answerer, so the waiting sender's offer
// would be dropped (p2pHandleSignal ignores a signal from an unknown peer). Best-effort, bounded
// to recent sessions; called from the SSE handler right after a device connects.
func (s *Service) RedeliverPendingTo(ctx context.Context, userID, deviceName string) {
	if deviceName == "" {
		return
	}
	rows, err := s.queries.ListPendingTransfersByReceiver(ctx, db.ListPendingTransfersByReceiverParams{
		UserID:       userID,
		ReceiverName: deviceName,
		CreatedAt:    s.now().Unix() - pendingTransferWindowSec,
	})
	if err != nil {
		slog.Error("redeliver pending transfers failed", "err", err, "user", userID, "device", deviceName)
		return
	}
	for _, row := range rows {
		sha := ""
		if row.FileSha256 != nil {
			sha = *row.FileSha256
		}
		view := SessionView{
			ID:         row.ID,
			SenderName: row.SenderName,
			FileName:   row.FileName,
			FileSize:   row.FileSize,
			FileSHA256: sha,
		}
		s.hub.SendTo(userID, deviceName, hub.Event{
			Type: "transfer:incoming",
			Data: map[string]any{
				"session":     view,
				"auto_accept": row.FileSize <= AutoAcceptThreshold,
			},
		})
	}
}

// Transition validates and applies a state change, then broadcasts transfer:state
// to both sender and receiver. When entering RELAY_ACTIVE, additionally emits
// transfer:relay_ready with the chunk/stream URLs (brief §5). When entering a
// terminal state, releases the relay session if any.
func (s *Service) Transition(
	ctx context.Context,
	userID, sessionID, target string,
	mode, reason string,
	bytes int64,
) error {
	sess, err := s.queries.GetTransferSession(ctx, sessionID)
	if err != nil {
		return ErrNotFound
	}
	if sess.UserID != userID {
		return ErrForbidden
	}
	if !IsValidTransition(sess.State, target) {
		return fmt.Errorf("%w: %s→%s", ErrInvalidTransition, sess.State, target)
	}

	// Reserve the relay slot BEFORE committing the state change: if the relay is
	// full (global or per-user cap) the transfer must not advance into a relay
	// mode it has no slot for — surface a retryable error instead. Registering
	// here also records the two legitimate participants, which is what later lets
	// the relay endpoints reject anyone who isn't the sender or receiver (R1/G2).
	if target == StateRelayActive {
		if err := s.relay.Register(sessionID, userID, sess.SenderName, sess.ReceiverName); err != nil {
			return fmt.Errorf("%w: %v", ErrRelayUnavailable, err)
		}
	}

	args := db.UpdateTransferStateParams{
		State: target,
		ID:    sessionID,
	}
	if mode != "" {
		v := mode
		args.Mode = &v
	}
	if reason != "" {
		v := reason
		args.FailReason = &v
	}
	if IsTerminal(target) {
		t := s.now().Unix()
		args.FinishedAt = &t
	}
	if bytes > 0 {
		v := bytes
		args.BytesTransferred = &v
	}

	if _, err := s.queries.UpdateTransferState(ctx, args); err != nil {
		return fmt.Errorf("update state: %w", err)
	}

	stateEv := hub.Event{
		Type: "transfer:state",
		Data: stateEventPayload(sessionID, target, mode, reason),
	}
	deliveredSender := s.hub.SendTo(userID, sess.SenderName, stateEv)
	deliveredReceiver := s.hub.SendTo(userID, sess.ReceiverName, stateEv)

	// Terminal completion/failure → notify any party whose page has closed.
	// Both Web Push and APNs are attempted; disjoint platform rows mean
	// firing both is always correct — each sender no-ops for the other's rows.
	if (target == StateDone || target == StateFailed) && (s.push.Enabled() || s.apns.Enabled()) {
		kind := push.KindTransferDone
		if target == StateFailed {
			kind = push.KindTransferFailed
		}
		n := push.Notification{
			Type:   kind,
			Params: map[string]string{"filename": sess.FileName},
			Tag:    "transfer:" + sessionID,
		}
		if !deliveredSender {
			if s.push.Enabled() {
				go s.push.Notify(context.Background(), userID, sess.SenderName, n)
			}
			if s.apns.Enabled() {
				go s.apns.Notify(context.Background(), userID, sess.SenderName, n)
			}
		}
		if !deliveredReceiver {
			if s.push.Enabled() {
				go s.push.Notify(context.Background(), userID, sess.ReceiverName, n)
			}
			if s.apns.Enabled() {
				go s.apns.Notify(context.Background(), userID, sess.ReceiverName, n)
			}
		}
	}

	if target == StateRelayActive {
		readyEv := hub.Event{
			Type: "transfer:relay_ready",
			Data: map[string]any{
				"session_id":   sessionID,
				"sender_url":   fmt.Sprintf("/api/relay/%s/chunk", sessionID),
				"receiver_url": fmt.Sprintf("/api/relay/%s/stream", sessionID),
			},
		}
		s.hub.SendTo(userID, sess.SenderName, readyEv)
		s.hub.SendTo(userID, sess.ReceiverName, readyEv)
	}

	if IsTerminal(target) {
		s.relay.Release(sessionID)
	}
	return nil
}

func stateEventPayload(sessionID, state, mode, reason string) map[string]any {
	p := map[string]any{
		"session_id": sessionID,
		"state":      state,
	}
	if mode != "" {
		p["mode"] = mode
	}
	if reason != "" {
		p["reason"] = reason
	}
	return p
}

// ExpirePending cancels any PENDING sessions older than now-ttl and pushes
// transfer:state to both peers. Returns the number of expired sessions.
func (s *Service) ExpirePending(ctx context.Context, ttl time.Duration) (int, error) {
	now := s.now()
	cutoff := now.Add(-ttl).Unix()
	finishedAt := now.Unix()

	rows, err := s.queries.ExpirePendingSessions(ctx, db.ExpirePendingSessionsParams{
		FinishedAt: &finishedAt,
		Cutoff:     cutoff,
	})
	if err != nil {
		return 0, fmt.Errorf("expire pending: %w", err)
	}
	for _, r := range rows {
		ev := hub.Event{
			Type: "transfer:state",
			Data: stateEventPayload(r.ID, StateCancelled, "", "pending_timeout"),
		}
		s.hub.SendTo(r.UserID, r.SenderName, ev)
		s.hub.SendTo(r.UserID, r.ReceiverName, ev)
		s.relay.Release(r.ID)
	}
	return len(rows), nil
}