Commilitia-Drop/internal/transfer/service.go

package transfer

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

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

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

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

func NewService(queries *db.Queries, h *hub.Hub, r RelayController) *Service {
	if r == nil {
		r = nopRelay{}
	}
	return &Service{
		queries: queries,
		hub:     h,
		relay:   r,
		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.
func (s *Service) Init(ctx context.Context, p InitParams) (string, error) {
	if p.SenderName == "" || p.ReceiverName == "" || p.FileName == "" {
		return "", errors.New("init: sender_name, receiver_name, file_name required")
	}
	if p.SenderName == p.ReceiverName {
		return "", 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 "", 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 "", 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

	s.hub.SendTo(p.UserID, p.ReceiverName, hub.Event{
		Type: "transfer:incoming",
		Data: map[string]any{
			"session":     view,
			"auto_accept": autoAccept,
		},
	})

	return id, nil
}

// 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),
	}
	s.hub.SendTo(userID, sess.SenderName, stateEv)
	s.hub.SendTo(userID, sess.ReceiverName, stateEv)

	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
}