rkcy

type AdminClient interface {
	GetMetadata(topic *string, allTopics bool, timeoutMs int) (*kafka.Metadata, error)
	CreateTopics(
		ctx context.Context,
		topics []kafka.TopicSpecification,
		options ...kafka.CreateTopicsAdminOption,
	) ([]kafka.TopicResult, error)
	Close()
}

type ClientCode struct {
	StorageInits    map[string]StorageInit
	ConcernHandlers ConcernHandlers
}

type ComplexConfigHandler interface {
	GetKey(proto.Message) string
	Unmarshal(b []byte) (proto.Message, error)
	UnmarshalJson(b []byte) (proto.Message, error)
}

type ConcernHandler interface {
	ConcernName() string
	HandleLogicCommand(
		ctx context.Context,
		system rkcypb.System,
		command string,
		direction rkcypb.Direction,
		args *StepArgs,
		instanceStore *InstanceStore,
		confRdr ConfigRdr,
	) (*rkcypb.ApecsTxn_Step_Result, []*rkcypb.ApecsTxn_Step)
	HandleCrudCommand(
		ctx context.Context,
		wg *sync.WaitGroup,
		system rkcypb.System,
		command string,
		direction rkcypb.Direction,
		args *StepArgs,
		storageType string,
	) (*rkcypb.ApecsTxn_Step_Result, []*rkcypb.ApecsTxn_Step)
	DecodeInstance(ctx context.Context, buffer []byte) (*ResultProto, error)
	DecodeArg(ctx context.Context, system rkcypb.System, command string, buffer []byte) (*ResultProto, error)
	DecodeResult(ctx context.Context, system rkcypb.System, command string, buffer []byte) (*ResultProto, error)
	DecodeRelatedRequest(ctx context.Context, relReq *rkcypb.RelatedRequest) (*ResultProto, error)

	SetLogicHandler(commands interface{}) error
	SetCrudHandler(storageType string, commands interface{}) error
	ValidateHandlers() bool
	SetStorageTargets(storageTargets map[string]*StorageTargetInit)
}

type ConcernHandlers map[string]ConcernHandler

type Config struct {
	rkcypb.Config
}

type ConfigRdr interface {
	GetString(key string) (string, bool)
	GetBool(key string) (bool, bool)
	GetFloat64(key string) (float64, bool)
	GetComplexMsg(msgType string, key string) (proto.Message, bool)
	GetComplexBytes(msgType string, key string) ([]byte, bool)

	BuildConfigResponse() *rkcypb.ConfigReadResponse
}

type Consumer interface {
	Assign(partitions []kafka.TopicPartition) error
	Close() error
	Commit() ([]kafka.TopicPartition, error)
	CommitOffsets(offsets []kafka.TopicPartition) ([]kafka.TopicPartition, error)
	QueryWatermarkOffsets(topic string, partition int32, timeoutMs int) (int64, int64, error)
	ReadMessage(timeout time.Duration) (*kafka.Message, error)
	StoreOffsets(offsets []kafka.TopicPartition) ([]kafka.TopicPartition, error)
}

type Error struct {
	Code rkcypb.Code
	Msg  string
}

type Fnv64HashFunc struct {
	// contains filtered or unexported fields
}

type HashFunc interface {
	Hash(val []byte, count int32) int32
}

type InstanceRecord struct {
	Instance   []byte
	Related    []byte
	CmpdOffset *rkcypb.CompoundOffset
	LastAccess time.Time
}

type InstanceStore struct {
	// contains filtered or unexported fields
}

type PlatformArgs struct {
	Id                string
	Platform          string
	Environment       string
	AdminBrokers      string
	AdminPingInterval time.Duration
}

type PlatformDiff struct {
	ProgsToStop  []*rkcypb.Program
	ProgsToStart []*rkcypb.Program
}

type Producer interface {
	Produce(msg *kafka.Message, deliveryChan chan kafka.Event) error
	Close()
	Events() chan kafka.Event
	Flush(timeoutMs int) int
}

type ProducerCh chan *kafka.Message

type ResultProto struct {
	Type     string
	Instance proto.Message
	Related  proto.Message
}

type RevertType int

type RtApecsTxn struct {
	Txn         *rkcypb.ApecsTxn
	TraceParent string
}

type RtConcern struct {
	Concern *rkcypb.Concern
	Topics  map[string]*RtTopics
}

type RtPlatformDef struct {
	PlatformDef          *rkcypb.PlatformDef
	Hash                 string
	Concerns             map[string]*RtConcern
	DefaultResponseTopic *RtTopics
	Clusters             map[string]*rkcypb.Cluster
	AdminCluster         *rkcypb.Cluster
	StorageTargets       map[string]*rkcypb.StorageTarget
	PrimaryStorageTarget string
}

type RtTopics struct {
	Topics                     *rkcypb.Concern_Topics
	CurrentTopic               string
	CurrentTopicPartitionCount int32
	CurrentCluster             *rkcypb.Cluster
	FutureTopic                string
	FutureTopicPartitionCount  int32
	FutureCluster              *rkcypb.Cluster
}

type StandardTopicName string

type Step struct {
	Concern       string
	Command       string
	Key           string
	Payload       proto.Message
	EffectiveTime time.Time
}

type StepArgs struct {
	TxnId         string
	Key           string
	Instance      []byte
	Payload       []byte
	EffectiveTime time.Time

	CmpdOffset    *rkcypb.CompoundOffset
	ForwardResult *rkcypb.ApecsTxn_Step_Result
}

type StorageInit func(ctx context.Context, wg *sync.WaitGroup, config map[string]string) error

type StorageTargetInit struct {
	*rkcypb.StorageTarget
	Init StorageInit
}

type StreamProvider interface {
	Type() string
	NewConsumer(brokers string, groupName string, logCh chan kafka.LogEvent) (Consumer, error)
	NewProducer(brokers string, logCh chan kafka.LogEvent) (Producer, error)
	NewAdminClient(brokers string) (AdminClient, error)
}

type TopicParts struct {
	FullTopic   string
	Platform    string
	Environment string
	Concern     string
	Topic       string
	System      rkcypb.System
	ConcernType rkcypb.Concern_Type
	Generation  int32
}

type Txn struct {
	Revert RevertType
	Steps  []Step
}