/// heavily inspired by <https://github.com/bluesky-social/atproto/blob/c7f5a868837d3e9b3289f988fee2267789327b06/packages/tap/README.md>

package tapc

import (
	"bytes"
	"context"
	"encoding/base64"
	"encoding/json"
	"fmt"
	"net/http"
	"net/url"
	"time"

	"github.com/bluesky-social/indigo/atproto/syntax"
	"github.com/gorilla/websocket"
	"tangled.org/core/log"
)

type Handler interface {
	OnEvent(ctx context.Context, evt Event) error
	OnError(ctx context.Context, err error)
}

type ConnectHandler interface {
	OnConnect(ctx context.Context)
}

type Client struct {
	Url           string
	AdminPassword string
	HTTPClient    *http.Client
}

func NewClient(url, adminPassword string) Client {
	return Client{
		Url:           url,
		AdminPassword: adminPassword,
		HTTPClient:    &http.Client{},
	}
}

// tap inserts one /repos/add body as a single sqlite statement at about 9
// variables per did, so a spindle's whole member list blows past sqlite's
// 32766 cap and tap 500s
const addReposBatchSize = 1000

func (c *Client) AddRepos(ctx context.Context, dids []syntax.DID) error {
	for len(dids) > 0 {
		n := min(len(dids), addReposBatchSize)
		if err := c.postDids(ctx, "/repos/add", dids[:n]); err != nil {
			return err
		}
		dids = dids[n:]
	}
	return nil
}

func (c *Client) RemoveRepos(ctx context.Context, dids []syntax.DID) error {
	return c.postDids(ctx, "/repos/remove", dids)
}

func (c *Client) postDids(ctx context.Context, path string, dids []syntax.DID) error {
	body, err := json.Marshal(map[string][]syntax.DID{"dids": dids})
	if err != nil {
		return err
	}
	req, err := http.NewRequestWithContext(ctx, "POST", c.Url+path, bytes.NewReader(body))
	if err != nil {
		return err
	}
	req.SetBasicAuth("admin", c.AdminPassword)
	req.Header.Set("Content-Type", "application/json")

	resp, err := c.HTTPClient.Do(req)
	if err != nil {
		return err
	}
	defer resp.Body.Close()
	if resp.StatusCode != http.StatusOK {
		return fmt.Errorf("tap: %s failed with status %d", path, resp.StatusCode)
	}
	return nil
}

func (c *Client) Connect(ctx context.Context, handler Handler) error {
	l := log.FromContext(ctx)

	u, err := url.Parse(c.Url)
	if err != nil {
		return err
	}
	if u.Scheme == "https" {
		u.Scheme = "wss"
	} else {
		u.Scheme = "ws"
	}
	u.Path = "/channel"

	url := u.String()
	basicAuth := "Basic " + base64.StdEncoding.EncodeToString([]byte("admin:"+c.AdminPassword))

	var backoff int
	for {
		select {
		case <-ctx.Done():
			return ctx.Err()
		default:
		}

		header := http.Header{
			"Authorization": []string{basicAuth},
		}
		conn, res, err := websocket.DefaultDialer.DialContext(ctx, url, header)
		if err != nil {
			if backoff < 12 {
				backoff++
			}
			l.Warn("dialing failed", "url", url, "err", err, "backoff", backoff)
			time.Sleep(time.Duration(5*backoff) * time.Second)

			continue
		}
		backoff = 0
		l.Info("connected to tap service", "subscription_code", res.StatusCode)

		if ch, ok := handler.(ConnectHandler); ok {
			ch.OnConnect(ctx)
		}

		if err = c.handleConnection(ctx, conn, handler); err != nil {
			l.Warn("tap connection failed", "err", err)
		}
	}
}

func (c *Client) handleConnection(ctx context.Context, conn *websocket.Conn, handler Handler) error {
	l := log.FromContext(ctx)

	defer func() {
		conn.Close()
		l.Warn("closed tap connection")
	}()
	l.Info("established tap connection")

	for {
		select {
		case <-ctx.Done():
			return ctx.Err()
		default:
		}
		_, message, err := conn.ReadMessage()
		if err != nil {
			return err
		}

		var ev Event
		if err := json.Unmarshal(message, &ev); err != nil {
			handler.OnError(ctx, fmt.Errorf("failed to parse message: %w", err))
			continue
		}
		if err := handler.OnEvent(ctx, ev); err != nil {
			handler.OnError(ctx, fmt.Errorf("failed to process event %d: %w", ev.ID, err))
			continue
		}

		ack := map[string]any{
			"type": "ack",
			"id":   ev.ID,
		}
		if err := conn.WriteJSON(ack); err != nil {
			l.Warn("failed to send ack", "err", err)
			continue
		}
	}
}