package main

import (
    "context"
    "log/slog"

    "github.com/ochaton/lymbo"
    "github.com/ochaton/lymbo/store/memory"
)

func main() {
    ctx := context.Background()

    kh := lymbo.NewKharon(memory.NewStore(), lymbo.DefaultSettings(), slog.Default())

    router := lymbo.NewRouter()
    router.HandleFunc("example", func(ctx context.Context, t *lymbo.Ticket) error {
        slog.Info("processing", "id", t.ID, "payload", t.Payload)
        return kh.Ack(ctx, t.ID)
    })

    go kh.Run(ctx, router)

    ticket, _ := lymbo.NewTicket(lymbo.MustID(), "example")
    kh.Put(ctx, *ticket)
}

ticket, err := lymbo.NewTicket(lymbo.MustID(), "task-type")

// With payload (auto-marshaled to JSON)
ticket.WithPayload(map[string]any{"key": "value"})

// With priority and delayed execution
ticket.WithNice(5).WithRunat(time.Now().Add(time.Hour))

// With tube routing
ticket, err := lymbo.NewTubeTicket("emails", lymbo.MustID(), "send-email")

// Put with options
kh.Put(ctx, *ticket,
    lymbo.WithDelay(lymbo.FixedDelay(5*time.Minute)),
    lymbo.WithNice(10),
)

router := lymbo.NewRouter()

router.HandleFunc("email", func(ctx context.Context, t *lymbo.Ticket) error {
    if err := sendEmail(t.Payload); err != nil {
        if isTransient(err) {
            return kh.Retry(ctx, t.ID, lymbo.WithDelay(lymbo.FixedDelay(5*time.Minute)))
        }
        return kh.Fail(ctx, t.ID, lymbo.WithErrorReason(err.Error()))
    }
    return kh.Ack(ctx, t.ID)
})

router.NotFoundFunc(func(ctx context.Context, t *lymbo.Ticket) error {
    return kh.Fail(ctx, t.ID, lymbo.WithErrorReason("unsupported type"))
})

// Retry with exponential backoff
kh.Retry(ctx, id, lymbo.WithDelay(lymbo.BackoffDelay(1.5, 15*time.Second, 0)))

// Done with TTL (auto-removed after 24h)
kh.Done(ctx, id, lymbo.WithDelay(lymbo.FixedDelay(24*time.Hour)))

// Fail with error reason
kh.Fail(ctx, id,
    lymbo.WithErrorReason("connection timeout"),
    lymbo.WithDelay(lymbo.FixedDelay(7*24*time.Hour)),
)

// Cancel but keep for audit
kh.Cancel(ctx, id, lymbo.WithKeep(), lymbo.WithErrorReason("cancelled by user"))

// Fixed delay
lymbo.FixedDelay(5 * time.Minute)

// Exponential backoff: base^attempts seconds, capped at maxDelay, with optional jitter
lymbo.BackoffDelay(1.5, 15*time.Second, 0)
lymbo.BackoffDelay(2.0, time.Minute, 500*time.Millisecond)

// Create a group handle (lightweight, no DB write)
g := kh.Group("order-42-notifications")

// Submit tickets into the group
for _, userID := range recipients {
    ticket, _ := lymbo.NewTicket(lymbo.MustID(), "send-notification")
    kh.Put(ctx, *ticket,
        lymbo.WithGroup(g.ID()),
        lymbo.WithPayload(userID),
    )
}

// Poll group progress from anywhere in your code
n, err := g.PendingCount(ctx)   // tickets still pending
ok, err := g.AllTerminal(ctx)   // true when none are pending

ticket.WithGroup("order-42-notifications")
kh.Put(ctx, *ticket)

// Move a ticket to a different group on retry
kh.Retry(ctx, t.ID,
    lymbo.WithGroup("retry-batch"),
    lymbo.WithDelay(lymbo.FixedDelay(5*time.Minute)),
)

// Clear the group by moving to an empty-string group is not supported;
// use a dedicated "archived" group name instead.

groupID := "order-42-notifications"

for _, userID := range recipients {
    ticket, _ := lymbo.NewTicket(lymbo.MustID(), "send-notification")
    kh.Put(ctx, *ticket, lymbo.WithGroup(groupID), lymbo.WithPayload(userID))
}

finalizer, _ := lymbo.NewTicket(lymbo.MustID(), "notifications-complete")
kh.Put(ctx, *finalizer, lymbo.AfterGroup(groupID))

// Declaratively, with an initial subscription set:
kh := lymbo.NewKharon(store,
    lymbo.DefaultSettings().WithOnlyTubes("emails", "notifications"),
    slog.Default(),
)

// Or empty, to drive subscriptions dynamically at runtime:
kh := lymbo.NewKharon(store,
    lymbo.DefaultSettings().EnableTubes(),
    slog.Default(),
)

if err := kh.Subscribe([]string{"emails", "sms"}); err != nil {
    // returns lymbo.ErrTubesNotEnabled if the instance was built without
    // WithOnlyTubes/EnableTubes
}

if err := kh.Unsubscribe([]string{"sms"}); err != nil { /* ... */ }

current := kh.Tubes()                          // snapshot of subscribed tubes
ok := kh.IsSubscribedTo(lymbo.Tube("emails"))  // membership check

settings := lymbo.DefaultSettings().
    WithWorkers(10).
    WithProcessTime(5 * time.Minute).
    WithBackoffBase(2.0).
    WithOnlyTubes("emails", "notifications")

import "github.com/ochaton/lymbo/store/memory"

store := memory.NewStore()

import (
    "github.com/jackc/pgx/v5/pgxpool"
    "github.com/ochaton/lymbo/store/postgres"
)

pool, _ := pgxpool.New(ctx, "postgres://user:pass@localhost/dbname")

store := postgres.NewTicketsRepository(pool)
store.Migrate(ctx) // creates schema automatically

store, _ := postgres.NewTicketsRepositoryWithConfig(postgres.Config{
    TableName: "my_tickets",
    Pool:      pool,
})
store.Migrate(ctx)

type Store interface {
    Get(context.Context, TicketId) (Ticket, error)
    Put(context.Context, Ticket) error
    Delete(context.Context, TicketId) error
    Update(context.Context, TicketId, UpdateFunc) error
    DeleteBatch(ctx context.Context, ids []TicketId) ([]TransitionInfo, error)
    UpdateBatch(ctx context.Context, updates []UpdateSet) ([]TransitionInfo, error)
    PollPending(context.Context, PollRequest) (PollResult, error)
    ExpireTickets(ctx context.Context, limit int, now time.Time) ([]TransitionInfo, error)
    CountPendingInGroup(ctx context.Context, groupID string) (int, error)
    PutAfterGroup(ctx context.Context, ticket Ticket, groupID string) error
}

import (
    "github.com/prometheus/client_golang/prometheus"
    lymbo_prom "github.com/ochaton/lymbo/prometheus"
)

// Single instance, default registry:
prometheus.MustRegister(lymbo_prom.NewCollector(kh, nil))

// Multiple instances in the same process:
reg := prometheus.NewRegistry()
reg.MustRegister(lymbo_prom.NewCollector(khPinger, prometheus.Labels{"kharon": "pinger"}))
reg.MustRegister(lymbo_prom.NewCollector(khPonger, prometheus.Labels{"kharon": "ponger"}))