pubsubx

type Config struct {
	Scope     string          `json:"scope"`
	Provider  string          `json:"provider"`
	Providers ProvidersConfig `json:"providers"`
}

type Errors []error

type Handler func(ctx context.Context, msgs []*messagex.Message) ([]error, error)

type Handlers map[messagex.Topic]Handler

type InMemoryConfig struct{}

type KafkaConfig struct {
	Brokers []string `json:"brokers"`
}

type ProvidersConfig struct {
	InMemory InMemoryConfig `json:"inmemory"`
	Kafka    KafkaConfig    `json:"kafka"`
}

type PubSub interface {
	// Publisher returns a Publisher instance for publishing messages.
	Publisher() Publisher

	// Subscriber returns a Subscriber instance for subscribing to messages.
	// It takes a group name, a list of topics, and optional SubscriberOptions.
	// It returns a Subscriber and an error if any.
	// A subscriber should define ALL the topics it wants to subscribe to.
	Subscriber(group string, topics []messagex.Topic, opts ...SubscriberOption) (Subscriber, error)

	// PubSubAdminClient returns a PubSubAdminClient instance for managing topics and configurations.
	// A new client is always returned on each call. Caller is responsible for closing the client.
	AdminClient() (PubSubAdminClient, error)

	// Close closes all publishers and subscribers associated with the PubSub instance.
	// It returns an error if any.
	Close() error
}

type PubSubAdminClient interface {
	// CreateTopic creates a topic with the given configuration.
	// The default configuration entries are set by default, but they can be overridden (see `pubsub.NewCreateTopicConfigEntries()`).
	CreateTopic(ctx context.Context, partitions int32, replicationFactor int16, topic string, configs ...map[string]*string) (kadm.CreateTopicResponse, error)
	// DeleteTopic deletes a topic.
	DeleteTopic(ctx context.Context, topic string) (kadm.DeleteTopicResponse, error)
	// HealthCheck checks the health of the underlying pubsub. It returns an error if the pubsub is unhealthy or we cannot connect to it.
	HealthCheck(ctx context.Context) error

	// ListTopics returns the details of the given topics.
	// If no topics are provided, it returns the details of all topics.
	ListTopics(ctx context.Context, topics ...string) (kadm.TopicDetails, error)

	Close()
}

type PubSubOption func(*PubSubOptions)

type PubSubOptions struct {
	TracerProvider                  trace.TracerProvider
	Propagator                      propagation.TextMapPropagator
	MeterProvider                   metric.MeterProvider
	DefaultCreateTopicConfigEntries map[string]*string
}

type Publisher interface {
	// PublishSync publishes messages synchronously to the specified topic.
	// It returns an error if the operation fails.
	PublishSync(ctx context.Context, topic messagex.Topic, messages ...*messagex.Message) (Errors, error)

	// PublishAsync publishes messages asynchronously to the specified topic.
	// It returns an error if the operation fails.
	//
	// WARNING: The context should stay alive until the messages are published.
	// When using a fire-n-forget approach in a request-response scenario, a new Context.Background() should be preferred since
	// the request context might be canceled before the messages are actually published.
	PublishAsync(ctx context.Context, topic messagex.Topic, messages ...*messagex.Message) error

	// Close closes the publisher.
	// Once a publisher is closed, it cannot be used to publish messages anymore.
	//
	// WARNING: Since the PubSub.Publisher() method always returns the same instance of the publisher,
	// closing the publisher will force the user to create a new PubSub instance to get a new publisher.
	// This should really be done only when the application is shutting down, or you no longer have any needs to publish messages.
	Close() error
}

type Subscriber interface {
	// Subscribe subscribes to all topics that are configured in the subscriber.
	// It takes a context and a map of topic handlers as input.
	// - If there are topics missing handlers, it will return an error immediately.
	Subscribe(ctx context.Context, topicHandlers Handlers) error
	// Close closes the subscriber.
	Close() error
}

type SubscriberOption func(*SubscriberOptions)

type SubscriberOptions struct {
	// MaxBatchSize max amount of elements the batch will contain.
	// Default value is 100 if nothing is specified.
	MaxBatchSize uint16
}