feat: add master-follower replication and client failover

Followers replicate from the master via sequence-based gRPC
sync with automatic retry. Failover client tries peers in order
and verifies connectivity with a Status call before returning.

Signed-off-by: Murat Parlakisik <parlakisik@gmail.com>

Author: parlakisik

internal/hub/failover.go

@@ -0,0 +1,67 @@
+//   /    ctx:                         https://ctx.ist
+// ,'`./    do you remember?
+// `.,'\
+//   \    Copyright 2026-present Context contributors.
+//                 SPDX-License-Identifier: Apache-2.0
+
+package hub
+
+import (
+	"context"
+
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials/insecure"
+)
+
+// NewFailoverClient creates a client that tries peers in
+// order until one succeeds. The first address is the
+// primary; others are fallbacks.
+//
+// Parameters:
+//   - peers: ordered list of hub addresses
+//   - bearerToken: token for authenticated RPCs
+//
+// Returns:
+//   - *Client: connected client to the first reachable peer
+//   - error: non-nil if no peer is reachable
+func NewFailoverClient(
+	peers []string, bearerToken string,
+) (*Client, error) {
+	var lastErr error
+	for _, addr := range peers {
+		conn, dialErr := grpc.NewClient(
+			addr,
+			grpc.WithTransportCredentials(
+				insecure.NewCredentials(),
+			),
+			grpc.WithDefaultCallOptions(
+				grpc.CallContentSubtype(codecName),
+			),
+		)
+		if dialErr != nil {
+			lastErr = dialErr
+			continue
+		}
+
+		// Verify connectivity with a Status call.
+		resp := &StatusResponse{}
+		callErr := conn.Invoke(
+			addBearerMD(
+				context.Background(), bearerToken,
+			),
+			"/ctx.hub.v1.CtxHub/Status",
+			&struct{}{},
+			resp,
+		)
+		if callErr != nil {
+			_ = conn.Close()
+			lastErr = callErr
+			continue
+		}
+
+		return &Client{
+			conn: conn, token: bearerToken,
+		}, nil
+	}
+	return nil, lastErr
+}

internal/hub/replicate.go

@@ -0,0 +1,50 @@
+//   /    ctx:                         https://ctx.ist
+// ,'`./    do you remember?
+// `.,'\
+//   \    Copyright 2026-present Context contributors.
+//                 SPDX-License-Identifier: Apache-2.0
+
+package hub
+
+import (
+	"context"
+	"time"
+)
+
+// replicateInterval is how often a follower retries
+// connecting to the master for replication.
+const replicateInterval = 5 * time.Second
+
+// StartReplication connects to the master and streams
+// entries into the local store. Blocks until the context
+// is cancelled. Retries on failure.
+//
+// Parameters:
+//   - ctx: context for cancellation
+//   - masterAddr: gRPC address of the master hub
+//   - store: local store to write replicated entries
+//   - clientToken: bearer token for auth
+func StartReplication(
+	ctx context.Context,
+	masterAddr string,
+	store *Store,
+	clientToken string,
+) {
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		default:
+		}
+
+		replicateOnce(
+			ctx, masterAddr, store, clientToken,
+		)
+
+		select {
+		case <-ctx.Done():
+			return
+		case <-time.After(replicateInterval):
+		}
+	}
+}

internal/hub/sync_helper.go

@@ -0,0 +1,101 @@
+//   /    ctx:                         https://ctx.ist
+// ,'`./    do you remember?
+// `.,'\
+//   \    Copyright 2026-present Context contributors.
+//                 SPDX-License-Identifier: Apache-2.0
+
+package hub
+
+import (
+	"context"
+	"time"
+
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials/insecure"
+	"google.golang.org/grpc/metadata"
+)
+
+// replicateOnce connects to the master, syncs all entries
+// since the local store's last sequence, and appends them.
+func replicateOnce(
+	ctx context.Context,
+	masterAddr string,
+	store *Store,
+	clientToken string,
+) {
+	conn, dialErr := grpc.NewClient(
+		masterAddr,
+		grpc.WithTransportCredentials(
+			insecure.NewCredentials(),
+		),
+		grpc.WithDefaultCallOptions(
+			grpc.CallContentSubtype(codecName),
+		),
+	)
+	if dialErr != nil {
+		return
+	}
+	defer func() { _ = conn.Close() }()
+
+	_, lastSeq := lastSequence(store)
+	authed := addBearerMD(ctx, clientToken)
+
+	stream, streamErr := conn.NewStream(
+		authed,
+		&grpc.StreamDesc{ServerStreams: true},
+		"/ctx.hub.v1.CtxHub/Sync",
+	)
+	if streamErr != nil {
+		return
+	}
+
+	if sendErr := stream.SendMsg(&SyncRequest{
+		SinceSequence: lastSeq,
+	}); sendErr != nil {
+		return
+	}
+	if closeErr := stream.CloseSend(); closeErr != nil {
+		return
+	}
+
+	for {
+		msg := &EntryMsg{}
+		if recvErr := stream.RecvMsg(msg); recvErr != nil {
+			return
+		}
+		entry := Entry{
+			ID:        msg.ID,
+			Type:      msg.Type,
+			Content:   msg.Content,
+			Origin:    msg.Origin,
+			Author:    msg.Author,
+			Timestamp: time.Unix(msg.Timestamp, 0),
+			Sequence:  msg.Sequence,
+		}
+		_, _ = store.Append([]Entry{entry})
+	}
+}
+
+// lastSequence returns the highest sequence in the store.
+func lastSequence(store *Store) (bool, uint64) {
+	all := store.Query(nil, 0)
+	if len(all) == 0 {
+		return false, 0
+	}
+	return true, all[len(all)-1].Sequence
+}
+
+// addBearerMD adds a bearer token to outgoing gRPC
+// metadata.
+func addBearerMD(
+	ctx context.Context, tok string,
+) context.Context {
+	if tok == "" {
+		return ctx
+	}
+	return metadata.NewOutgoingContext(
+		ctx, metadata.Pairs(
+			"authorization", bearerPrefix+tok,
+		),
+	)
+}