package main

import (
	"encoding/json"
)

type TestFilter struct {
}

type Config struct {
}

func New(b []byte) (interface{}, error) {
	var cfg Config
	if err := json.Unmarshal(b, &cfg); err != nil {
		return nil, err
	}

	return &TestFilter{}, nil
}

func (f *TestFilter) ID() string {
	return "testfilter"
}

func (f *TestFilter) IsAllowed(msg []byte) bool {
	return false
}

package main

import (
	"context"
	"fmt"
	"log/slog"
	"strconv"

	"github.com/redis/go-redis/v9"
	"github.com/zerodha/kaf-relay/pkg/relay"
)

const watermarkKey = "_watermark" // Redis hash: topic -> partition -> offset

// RedisTarget implements relay.Target, writing messages to Redis streams
// and tracking offsets in a Redis hash for resumption.
type RedisTarget struct {
	client *redis.Client
	topic  string
	log    *slog.Logger
	stopCh chan struct{}
	doneCh chan struct{}
}

func NewRedisTarget(addr, topic string, log *slog.Logger) *RedisTarget {
	return &RedisTarget{
		client: redis.NewClient(&redis.Options{Addr: addr}),
		topic:  topic,
		log:    log,
		stopCh: make(chan struct{}),
		doneCh: make(chan struct{}),
	}
}

// GetHighWatermark reads persisted offsets from a Redis hash for the relay to resume from the last checkpoint.
func (r *RedisTarget) GetHighWatermark(ctx context.Context) (relay.Offsets, error) {
	vals, err := r.client.HGetAll(ctx, watermarkKey).Result()
	if err != nil {
		return nil, err
	}

	offsets := make(map[int32]int64, len(vals))
	for field, val := range vals {
		p, _ := strconv.Atoi(field)
		o, _ := strconv.ParseInt(val, 10, 64)
		offsets[int32(p)] = o
	}

	return relay.Offsets{r.topic: offsets}, nil
}

// Start blocks until Close() is called.
func (r *RedisTarget) Start() error {
	defer close(r.doneCh)
	<-r.stopCh
	return nil
}

// Write writes a message to Redis and updates the watermark offset.
func (r *RedisTarget) Write(ctx context.Context, msg relay.Message) error {
	pipe := r.client.Pipeline()

	pipe.XAdd(ctx, &redis.XAddArgs{
		Stream: msg.Topic,
		Values: map[string]interface{}{
			"key":   string(msg.Key),
			"value": string(msg.Value),
		},
	})

	// Persist the source offset for GetHighWatermark() (for relay resumption).
	pipe.HSet(ctx, watermarkKey, fmt.Sprintf("%d", msg.Partition), msg.Offset)

	_, err := pipe.Exec(ctx)
	return err
}

// Close signals Start() to return and waits for it and closes the Redis client.
func (r *RedisTarget) Close() error {
	close(r.stopCh)
	<-r.doneCh
	return r.client.Close()
}

target := NewRedisTarget("localhost:6379", "my-stream", log)

srcPool, _ := relay.NewSourcePool(poolCfg, consumerCfgs, topic, nil, metricsSet, log)

r, _ := relay.NewRelay(relayCfg, srcPool, target, topic, nil, metricsSet, log)
r.Start(ctx)