package mysql

import (
	"bytes"
	"context"
	"database/sql"
	"encoding/base64"
	"encoding/json"
	"time"

	"git.yuntuo.hk/k2safe/orbit/internal/fault"
	"git.yuntuo.hk/k2safe/orbit/internal/jsonvalue"
	core "git.yuntuo.hk/k2safe/orbit/model"
	"git.yuntuo.hk/k2safe/orbit/platform/auth"
	"git.yuntuo.hk/k2safe/orbit/platform/events"
	platform "git.yuntuo.hk/k2safe/orbit/platform/model"
)

type eventCursor struct {
	Version    int                    `json:"v"`
	Kind       string                 `json:"k"`
	Namespace  core.ID                `json:"n"`
	Credential core.ID                `json:"c"`
	Parent     core.ID                `json:"p"`
	State      platform.DeliveryState `json:"s"`
	At         time.Time              `json:"t"`
	ID         core.ID                `json:"i"`
}

func parseEventCursor(raw string, t auth.Token, parent core.ID, kind string, state platform.DeliveryState) (eventCursor, error) {
	var c eventCursor
	b, err := base64.RawURLEncoding.Strict().DecodeString(raw)
	if err != nil || len(b) > 1024 {
		return c, fault.Invalid("invalid event cursor")
	}
	if _, err = jsonvalue.Canonical(b); err != nil {
		return c, fault.Invalid("invalid event cursor")
	}
	var fields map[string]json.RawMessage
	if json.Unmarshal(b, &fields) != nil || len(fields) != 8 {
		return c, fault.Invalid("invalid event cursor")
	}
	for _, k := range []string{"v", "k", "n", "c", "p", "s", "t", "i"} {
		if _, ok := fields[k]; !ok {
			return c, fault.Invalid("invalid event cursor")
		}
	}
	d := json.NewDecoder(bytes.NewReader(b))
	d.DisallowUnknownFields()
	if d.Decode(&c) != nil || c.Version != 1 || c.Kind != kind || c.Namespace != t.NamespaceID || c.Credential != t.CredentialID || c.Parent != parent || c.State != state || c.ID.Validate() != nil || c.At.Year() < 2000 || c.At.Year() > 9998 || c.At.Nanosecond()%1000 != 0 {
		return c, fault.Invalid("cursor does not match event query")
	}
	return c, nil
}
func makeEventCursor(t auth.Token, parent core.ID, kind string, state platform.DeliveryState, at time.Time, id core.ID) (string, error) {
	b, err := json.Marshal(eventCursor{1, kind, t.NamespaceID, t.CredentialID, parent, state, at, id})
	return base64.RawURLEncoding.EncodeToString(b), err
}
func (s *EventStore) AuthorizeEvents(ctx context.Context, t auth.Token) error {
	return s.transaction(ctx, t, "events:consume", func(*sql.Tx, auth.Permissions) error { return nil })
}
func (s *EventStore) BindConsumer(ctx context.Context, t auth.Token, id core.ID) (events.ConsumerBinding, error) {
	var out events.ConsumerBinding
	err := s.transaction(ctx, t, "events:consume", func(tx *sql.Tx, p auth.Permissions) error {
		sub, err := readSubscription(ctx, tx, t, p, id, false)
		if err != nil {
			return err
		}
		out.SubscriptionID = sub.ID
		out.QueueID = sub.QueueID
		out.Enabled = sub.Enabled
		return tx.QueryRowContext(ctx, `SELECT application_id,UTC_TIMESTAMP(6) FROM orbit_namespaces WHERE id=?`, t.NamespaceID).Scan(&out.ApplicationID, &out.ServerTime)
	})
	return out, err
}
func (s *EventStore) ListSubscriptions(ctx context.Context, t auth.Token, queue core.ID, in events.PageInput) (events.SubscriptionPage, error) {
	out := events.SubscriptionPage{Items: []events.Subscription{}}
	err := s.transaction(ctx, t, "events:read", func(tx *sql.Tx, p auth.Permissions) error {
		if !p.AllowsQueue(queue) {
			return fault.Forbidden
		}
		q := `SELECT ` + subscriptionColumns + ` FROM orbit_subscriptions WHERE namespace_id=? AND credential_id=? AND queue_id=?`
		args := []any{t.NamespaceID, t.CredentialID, queue}
		if t.Console != nil {
			q = `SELECT ` + subscriptionColumns + ` FROM orbit_subscriptions WHERE namespace_id=? AND queue_id=?`
			args = []any{t.NamespaceID, queue}
		}
		if in.Cursor != "" {
			c, err := parseEventCursor(in.Cursor, t, queue, "subscriptions", "")
			if err != nil {
				return err
			}
			q += ` AND (created_at<? OR (created_at=? AND id<?))`
			args = append(args, c.At, c.At, c.ID)
		}
		q += ` ORDER BY created_at DESC,id DESC LIMIT ?`
		args = append(args, in.Limit+1)
		rows, err := tx.QueryContext(ctx, q, args...)
		if err != nil {
			return err
		}
		defer rows.Close()
		for rows.Next() {
			v, err := scanSubscription(rows)
			if err != nil {
				return err
			}
			out.Items = append(out.Items, v)
		}
		if err = rows.Err(); err != nil {
			return err
		}
		if len(out.Items) > in.Limit {
			out.Items = out.Items[:in.Limit]
			last := out.Items[len(out.Items)-1]
			out.NextCursor, err = makeEventCursor(t, queue, "subscriptions", "", last.CreatedAt, last.ID)
		}
		return err
	})
	return out, err
}
func (s *EventStore) ListDeliveries(ctx context.Context, t auth.Token, id core.ID, in events.PageInput) (events.DeliveryPage, error) {
	out := events.DeliveryPage{Items: []events.Delivery{}}
	err := s.transaction(ctx, t, "events:read", func(tx *sql.Tx, p auth.Permissions) error {
		if _, err := readSubscription(ctx, tx, t, p, id, false); err != nil {
			return err
		}
		q := `SELECT d.id,d.subscription_id,d.event_id,r.sequence,d.revision,d.state,d.available_at,d.lease_epoch,d.lease_until,d.attempts,d.attempt_budget,d.acknowledged_at,d.created_at FROM orbit_deliveries d LEFT JOIN orbit_result_events r ON r.namespace_id=d.namespace_id AND r.id=d.event_id WHERE d.namespace_id=? AND d.subscription_id=?`
		args := []any{t.NamespaceID, id}
		if in.State != "" {
			q += ` AND d.state=?`
			args = append(args, in.State)
		}
		if in.Cursor != "" {
			c, err := parseEventCursor(in.Cursor, t, id, "deliveries", in.State)
			if err != nil {
				return err
			}
			q += ` AND (d.created_at<? OR (d.created_at=? AND d.id<?))`
			args = append(args, c.At, c.At, c.ID)
		}
		q += ` ORDER BY d.created_at DESC,d.id DESC LIMIT ?`
		args = append(args, in.Limit+1)
		rows, err := tx.QueryContext(ctx, q, args...)
		if err != nil {
			return err
		}
		defer rows.Close()
		for rows.Next() {
			var v events.Delivery
			if err = rows.Scan(&v.ID, &v.SubscriptionID, &v.EventID, &v.Sequence, &v.Revision, &v.State, &v.AvailableAt, &v.LeaseEpoch, &v.LeaseUntil, &v.Attempts, &v.AttemptBudget, &v.AcknowledgedAt, &v.CreatedAt); err != nil {
				return err
			}
			out.Items = append(out.Items, v)
		}
		if err = rows.Err(); err != nil {
			return err
		}
		if len(out.Items) > in.Limit {
			out.Items = out.Items[:in.Limit]
			last := out.Items[len(out.Items)-1]
			out.NextCursor, err = makeEventCursor(t, id, "deliveries", in.State, last.CreatedAt, last.ID)
		}
		return err
	})
	return out, err
}
func (s *EventStore) GetDeliveryDetail(ctx context.Context, t auth.Token, sub, id core.ID) (events.DeliveryDetail, error) {
	var out events.DeliveryDetail
	err := s.transaction(ctx, t, "events:read", func(tx *sql.Tx, p auth.Permissions) error {
		if _, err := readSubscription(ctx, tx, t, p, sub, false); err != nil {
			return err
		}
		d, err := readDelivery(ctx, tx, t.NamespaceID, sub, id, false)
		if err != nil {
			return err
		}
		out.Delivery = d.view
		out.Event, err = readResultEvent(ctx, tx, t.NamespaceID, d.view.EventID)
		return err
	})
	return out, err
}