package appservice

import (
	"context"
	"encoding/json"
	"errors"
	"sync"
	"time"

	"github.com/dariuszjaniak/warp/server/internal/matrix/event"
	"github.com/dariuszjaniak/warp/server/internal/roomserver"
	"github.com/dariuszjaniak/warp/server/internal/storage"
)

// transactionSize is the most events one transaction carries, which is
// also how far the stream is read ahead at a time. Synapse sends 100.
const transactionSize = 100

// The backoff after a failed transaction: it doubles from the first value
// to the second and stays there. A bridge that is down for maintenance is
// tried every few minutes, not hammered, and is caught up within a minute
// or so of coming back.
const (
	minBackoff = 2 * time.Second
	maxBackoff = 5 * time.Minute
)

// idleRecheck is how often an idle worker looks at the stream without
// being woken. Wake-ups come from the emit stage and are never missed in
// a running server; this is the safety net for one that is.
const idleRecheck = 30 * time.Second

// Status is what the admin API shows of one service's delivery.
type Status struct {
	ID              string
	URL             string
	SenderUserID    string
	StreamPos       int64
	TxnID           int64
	Failures        int
	LastError       string
	LastSuccessMS   int64
	LastFailureMS   int64
	RetryAtMS       int64
	EventsDelivered int64
}

// worker delivers to one service.
type worker struct {
	svc  *Service
	wake chan struct{}

	// retry cuts a backoff short: a service that has just answered a ping
	// is up, and making it wait out minutes of backoff helps nobody.
	retry chan struct{}

	mu     sync.Mutex
	status Status
}

func newWorker(svc *Service) *worker {
	return &worker{
		svc:   svc,
		wake:  make(chan struct{}, 1),
		retry: make(chan struct{}, 1),
		status: Status{
			ID:           svc.ID,
			URL:          svc.URL,
			SenderUserID: svc.SenderUserID(),
		},
	}
}

// poke wakes the worker without blocking; one pending wake-up is as good
// as many.
func (w *worker) poke() {
	select {
	case w.wake <- struct{}{}:
	default:
	}
}

// retryNow cuts the current backoff short, without blocking.
func (w *worker) retryNow() {
	select {
	case w.retry <- struct{}{}:
	default:
	}
}

func (w *worker) update(fn func(*Status)) {
	w.mu.Lock()
	defer w.mu.Unlock()
	fn(&w.status)
}

func (w *worker) snapshot() Status {
	w.mu.Lock()
	defer w.mu.Unlock()
	return w.status
}

// run is a worker's loop: read the stream from the cursor, keep what the
// service is interested in, send it as a transaction until the service
// takes it, move the cursor, repeat.
//
// The cursor is in the database, and so is the transaction in flight, so
// a restart resumes where it stopped — with the same transaction ID for
// the same events, if one was being sent — and nothing is lost or sent
// twice under two IDs.
func (a *API) run(ctx context.Context, w *worker) {
	defer a.wg.Done()
	log := a.log.With("appservice", w.svc.ID)

	state, ok := a.loadState(ctx, w)
	if !ok {
		return
	}

	var pending []json.RawMessage
	for ctx.Err() == nil {
		if state.TxnTo != 0 {
			if pending == nil {
				// A transaction left in flight by an earlier run: read its
				// events again.
				events, err := a.collect(ctx, w.svc, state.StreamPos, state.TxnTo)
				if err != nil {
					log.Warn("could not read the events of a transaction in flight", "txn_id", state.TxnID, "error", err)
					if !sleep(ctx, minBackoff) {
						return
					}
					continue
				}
				pending = events
			}
			if !a.deliver(ctx, w, state.TxnID, pending) {
				return
			}
			if err := a.store.CompleteTransaction(ctx, w.svc.ID, state.TxnTo); err != nil {
				log.Warn("could not record a delivered transaction", "txn_id", state.TxnID, "error", err)
				if !sleep(ctx, minBackoff) {
					return
				}
				continue
			}
			delivered := int64(len(pending))
			state.StreamPos, state.TxnTo, pending = state.TxnTo, 0, nil
			w.update(func(s *Status) {
				s.StreamPos, s.TxnID = state.StreamPos, state.TxnID
				s.EventsDelivered += delivered
			})
			continue
		}

		batch, err := a.rooms.StreamEvents(ctx, state.StreamPos, 0, transactionSize)
		if err != nil {
			log.Warn("could not read the event stream", "error", err)
			if !sleep(ctx, minBackoff) {
				return
			}
			continue
		}
		if len(batch) == 0 {
			select {
			case <-ctx.Done():
				return
			case <-w.wake:
			case <-time.After(idleRecheck):
			}
			continue
		}

		to := batch[len(batch)-1].Position
		events, err := a.interesting(ctx, w.svc, batch)
		if err != nil {
			log.Warn("could not decide which events concern the service", "error", err)
			if !sleep(ctx, minBackoff) {
				return
			}
			continue
		}
		if len(events) == 0 {
			// Nothing here for the service: move past it without a
			// transaction.
			if err := a.store.CompleteTransaction(ctx, w.svc.ID, to); err != nil {
				log.Warn("could not advance the cursor", "error", err)
				if !sleep(ctx, minBackoff) {
					return
				}
				continue
			}
			state.StreamPos = to
			w.update(func(s *Status) { s.StreamPos = to })
			continue
		}

		txnID := state.TxnID + 1
		if err := a.store.BeginTransaction(ctx, w.svc.ID, txnID, to); err != nil {
			log.Warn("could not record a transaction", "error", err)
			if !sleep(ctx, minBackoff) {
				return
			}
			continue
		}
		state.TxnID, state.TxnTo, pending = txnID, to, events
	}
}

// loadState reads (or starts) the service's cursor. A service seen for the
// first time starts at the top of the stream: it is told what happens from
// now on, not the server's whole history.
func (a *API) loadState(ctx context.Context, w *worker) (storage.AppServiceState, bool) {
	for {
		top, err := a.rooms.StreamTop(ctx)
		if err == nil {
			var state storage.AppServiceState
			if state, err = a.store.AppServiceState(ctx, w.svc.ID, top); err == nil {
				w.update(func(s *Status) { s.StreamPos, s.TxnID = state.StreamPos, state.TxnID })
				return state, true
			}
		}
		a.log.Warn("could not read an application service's cursor", "appservice", w.svc.ID, "error", err)
		if !sleep(ctx, minBackoff) {
			return storage.AppServiceState{}, false
		}
	}
}

// collect reads the events of (from, to] the service is interested in.
func (a *API) collect(ctx context.Context, svc *Service, from, to int64) ([]json.RawMessage, error) {
	var out []json.RawMessage
	for from < to {
		batch, err := a.rooms.StreamEvents(ctx, from, to, transactionSize)
		if err != nil {
			return nil, err
		}
		if len(batch) == 0 {
			break
		}
		events, err := a.interesting(ctx, svc, batch)
		if err != nil {
			return nil, err
		}
		out = append(out, events...)
		from = batch[len(batch)-1].Position
	}
	if out == nil {
		out = []json.RawMessage{}
	}
	return out, nil
}

// interesting keeps the events of a batch that concern a service, encoded
// for a transaction.
//
// An event concerns a service when its sender or (for a membership) its
// target is one of the service's users, when its room is in the service's
// room namespaces, or when its room has one of the service's users in it
// or one of the service's aliases pointing at it. The last two are the
// rules that make a bridge see the conversations its puppets are in, and
// they are read once per room per batch, as the room is now: an event
// read late, from a backlog, is judged by who is in the room when it is
// read rather than when it happened.
func (a *API) interesting(ctx context.Context, svc *Service, batch []roomserver.StreamEvent) ([]json.RawMessage, error) {
	rooms := map[string]bool{}
	var out []json.RawMessage
	for _, se := range batch {
		e := se.Event
		keep := svc.InterestedInUser(e.Sender) || svc.InterestedInRoom(e.RoomID)
		if !keep && e.Type == event.TypeMember && e.StateKey != nil {
			keep = svc.InterestedInUser(*e.StateKey)
		}
		if !keep {
			roomInterest, seen := rooms[e.RoomID]
			if !seen {
				var err error
				if roomInterest, err = a.roomConcerns(ctx, svc, e.RoomID); err != nil {
					return nil, err
				}
				rooms[e.RoomID] = roomInterest
			}
			keep = roomInterest
		}
		if !keep {
			continue
		}
		raw, err := json.Marshal(e)
		if err != nil {
			return nil, err
		}
		out = append(out, raw)
	}
	return out, nil
}

// roomConcerns reports whether a room has one of a service's users in it
// or one of its aliases pointing at it.
func (a *API) roomConcerns(ctx context.Context, svc *Service, roomID string) (bool, error) {
	members, err := a.rooms.JoinedUserIDs(ctx, roomID)
	if err != nil && !errors.Is(err, storage.ErrNoSuchRoom) {
		return false, err
	}
	for _, m := range members {
		if svc.OwnsUser(m) {
			return true, nil
		}
	}
	if len(svc.Namespaces.Aliases) == 0 {
		return false, nil
	}
	aliases, err := a.rooms.LocalAliases(ctx, roomID)
	if err != nil && !errors.Is(err, storage.ErrNoSuchRoom) {
		return false, err
	}
	for _, alias := range aliases {
		if svc.OwnsAlias(alias) {
			return true, nil
		}
	}
	return false, nil
}

// deliver sends one transaction until the service accepts it. It returns
// false only when the worker is being stopped.
func (a *API) deliver(ctx context.Context, w *worker, txnID int64, events []json.RawMessage) bool {
	backoff := minBackoff
	for {
		err := a.client.SendTransaction(ctx, w.svc, txnID, events)
		now := a.now().UnixMilli()
		if err == nil {
			w.update(func(s *Status) {
				s.Failures, s.LastError, s.RetryAtMS, s.LastSuccessMS = 0, "", 0, now
			})
			return true
		}
		if ctx.Err() != nil {
			return false
		}
		w.update(func(s *Status) {
			s.Failures++
			s.LastError = err.Error()
			s.LastFailureMS = now
			s.RetryAtMS = now + backoff.Milliseconds()
		})
		a.log.Warn("could not deliver a transaction to an application service",
			"appservice", w.svc.ID, "txn_id", txnID, "events", len(events), "retry_in", backoff, "error", err)
		select {
		case <-ctx.Done():
			return false
		case <-time.After(backoff):
		case <-w.retry:
		}
		backoff = min(backoff*2, maxBackoff)
	}
}

// sleep waits for d or for ctx, reporting whether the wait ran its course.
func sleep(ctx context.Context, d time.Duration) bool {
	select {
	case <-ctx.Done():
		return false
	case <-time.After(d):
		return true
	}
}