cluster/store: add state recovery

internal/cluster/events/build.go

@@ -73,6 +73,9 @@ func (b *Builder) NewRecord() arrow.Record {
 func (b *Builder) Release() {
 	b.r.Release()
 }
+func (b *Builder) Len() int {
+	return b.r.Field(0).Len()
+}
 
 func (b *Builder) Write(list *v1.Data_List) arrow.Record {
 	ls := list.GetItems()
@@ -81,12 +84,12 @@ func (b *Builder) Write(list *v1.Data_List) arrow.Record {
 	})
 	b.r.Reserve(len(ls))
 	for _, e := range ls {
-		b.writeData(e)
+		b.WriteData(e)
 	}
 	return b.NewRecord()
 }
 
-func (b *Builder) writeData(data *v1.Data) {
+func (b *Builder) WriteData(data *v1.Data) {
 	fs := data.ProtoReflect()
 	fds := fs.Descriptor().Fields()
 	for i := 0; i < fds.Len(); i++ {

internal/cluster/http/service.go

@@ -608,6 +608,12 @@ func (s *Service) handleApiEvent(w http.ResponseWriter, r *http.Request, params
 
 func (s *Service) process(w http.ResponseWriter, r *http.Request, e *v1.Data) {
 	ctx := r.Context()
+	tenantId := s.tenants.TenantBySiteID(ctx, e.Domain)
+	if tenantId == "" {
+		w.WriteHeader(http.StatusUnauthorized)
+		return
+	}
+	e.TenantId = tenantId
 	err := s.store.Data(ctx, e)
 	if err == nil {
 		w.WriteHeader(http.StatusAccepted)

internal/cluster/snapshots/badger.go

@@ -0,0 +1,23 @@
+package snapshots
+
+import (
+	"github.com/dgraph-io/badger/v4"
+	"github.com/hashicorp/raft"
+)
+
+type Badger struct {
+	DB *badger.DB
+}
+
+func NewBadger(db *badger.DB) *Badger {
+	return &Badger{DB: db}
+}
+
+var _ raft.FSMSnapshot = (*Badger)(nil)
+
+func (b *Badger) Persist(sink raft.SnapshotSink) error {
+	_, err := b.DB.Backup(sink, 0)
+	return err
+}
+
+func (Badger) Release() {}