package keeper import ( "fmt" sdk "github.com/cosmos/cosmos-sdk/types" "github.com/oy/openyield/x/bearers/types" ) // msg_server.go implements the bearers module's MsgServer (G-023 ownership // split: cosmos-engineer scaffolds the file structure + method signatures; // mesh-engineer/backend-engineer implements the handler logic bodies). The // MsgServer wraps the Keeper + the BreadKeeper expected-keeper shim (already // on the Keeper). // // Each method returns a (*Response, error). Handler state-machine ordering // is enforced: ValidateBasic → keeper authz → state mutation → // ctx.EventManager().EmitEvent. // // Surveillance-resistant invariant (A-522): NO handler emits geolocation or // sender physical location. The surveillance-resistant locked const on // OYSATLink/OYLRLink is a runtime invariant — a handler that emits // geolocation violates it. A negative simtest asserts the event set // contains NO geolocation fields. // // One-shot OY-QR (A-521): the MsgConsumeOYQR handler flips consumed BEFORE // the transfer effect (state write FIRST, then the BreadKeeper shim call). // A replay finds consumed==true and returns an error (idempotent reject, // NOT double-effect). The SDK store is atomic per tx — a panic in the // transfer rolls back the whole tx, so the order is safe; the order // documents intent and matches the ibc-go delete-before-mint convention. // msgServer is the concrete MsgServer implementation wrapping the Keeper. type msgServer struct { Keeper } // NewMsgServerImpl returns the bearers MsgServer for the provided Keeper. func NewMsgServerImpl(k Keeper) types.MsgServer { return &msgServer{Keeper: k} } var _ types.MsgServer = msgServer{} // unwrapCtx extracts the sdk.Context from the interface-typed ctx. func unwrapCtx(ctx interface{}) sdk.Context { if c, ok := ctx.(sdk.Context); ok { return c } panic(fmt.Sprintf("bearers: expected sdk.Context, got %T", ctx)) } // nowUnix returns the current block time as unix seconds from the ctx. func nowUnix(ctx sdk.Context) int64 { return ctx.BlockTime().Unix() } // --- OpenSession (creates Session status=Open) ------------------------------- // OpenSession creates a new Session with status=Open. ValidateBasic is // stateless; the handler enforces idempotency (session-id must not already // exist). func (s msgServer) OpenSession(ctx interface{}, msg *types.MsgOpenSession) (*types.MsgOpenSessionResponse, error) { if err := msg.ValidateBasic(); err != nil { return nil, err } sdkCtx := unwrapCtx(ctx) // Idempotency: session-id must not already exist. if _, ok := s.Keeper.GetSession(sdkCtx, msg.SessionID); ok { return nil, fmt.Errorf("bearers: session %q already exists", msg.SessionID) } session := types.Session{ SessionID: msg.SessionID, BearerType: msg.BearerType, InitiatorReach: msg.InitiatorReach, PeerReach: msg.PeerReach, Status: types.SessionOpen, Frames: []types.Frame{}, TTL: msg.TTL, OpenedAt: nowUnix(sdkCtx), } s.Keeper.SetSession(sdkCtx, session) sdkCtx.EventManager().EmitEvent(sdk.NewEvent( "bearers.session_opened", sdk.NewAttribute("session_id", msg.SessionID), sdk.NewAttribute("bearer_type", string(msg.BearerType)), sdk.NewAttribute("initiator_reach", msg.InitiatorReach), sdk.NewAttribute("peer_reach", msg.PeerReach), sdk.NewAttribute("status", string(types.SessionOpen)), // NO geolocation (A-522 surveillance-resistant invariant). )) return &types.MsgOpenSessionResponse{}, nil } // --- CloseSession (Active → Closed) ------------------------------------------ // CloseSession transitions an Active session to Closed. The handler // enforces the stateful source-status check (must be Open or Active; an // Open session with no frames can close directly). func (s msgServer) CloseSession(ctx interface{}, msg *types.MsgCloseSession) (*types.MsgCloseSessionResponse, error) { if err := msg.ValidateBasic(); err != nil { return nil, err } sdkCtx := unwrapCtx(ctx) session, ok := s.Keeper.GetSession(sdkCtx, msg.SessionID) if !ok { return nil, fmt.Errorf("bearers: session %q not found", msg.SessionID) } if session.IsTerminal() { return nil, fmt.Errorf("bearers: session %q is terminal (%s), cannot close", msg.SessionID, session.Status) } session.Status = types.SessionClosed session.ClosedAt = nowUnix(sdkCtx) s.Keeper.SetSession(sdkCtx, session) sdkCtx.EventManager().EmitEvent(sdk.NewEvent( "bearers.session_closed", sdk.NewAttribute("session_id", msg.SessionID), sdk.NewAttribute("status", string(types.SessionClosed)), )) return &types.MsgCloseSessionResponse{}, nil } // --- RevokeSession (out-of-band → Revoked) ----------------------------------- // RevokeSession transitions a session to Revoked (out-of-band termination). // A revoked session rejects further Receive. The handler enforces the // stateful source-status check (must not already be terminal). func (s msgServer) RevokeSession(ctx interface{}, msg *types.MsgRevokeSession) (*types.MsgRevokeSessionResponse, error) { if err := msg.ValidateBasic(); err != nil { return nil, err } sdkCtx := unwrapCtx(ctx) session, ok := s.Keeper.GetSession(sdkCtx, msg.SessionID) if !ok { return nil, fmt.Errorf("bearers: session %q not found", msg.SessionID) } if session.IsTerminal() { return nil, fmt.Errorf("bearers: session %q is terminal (%s), cannot revoke", msg.SessionID, session.Status) } session.Status = types.SessionRevoked session.ClosedAt = nowUnix(sdkCtx) s.Keeper.SetSession(sdkCtx, session) sdkCtx.EventManager().EmitEvent(sdk.NewEvent( "bearers.session_revoked", sdk.NewAttribute("session_id", msg.SessionID), sdk.NewAttribute("status", string(types.SessionRevoked)), )) return &types.MsgRevokeSessionResponse{}, nil } // --- SendOYSATFrame (send a frame on an Open/Active session) ------------------ // SendOYSATFrame sends a frame on an OY-SAT session. The handler enforces // the stateful session-status check: the session must be Open or Active // (frames on Closed/Revoked are REJECTED — the rejected-frame case). func (s msgServer) SendOYSATFrame(ctx interface{}, msg *types.MsgSendOYSATFrame) (*types.MsgSendOYSATFrameResponse, error) { if err := msg.ValidateBasic(); err != nil { return nil, err } sdkCtx := unwrapCtx(ctx) session, ok := s.Keeper.GetSession(sdkCtx, msg.SessionID) if !ok { return nil, fmt.Errorf("bearers: session %q not found", msg.SessionID) } if session.IsTerminal() { // Rejected-frame case: a frame received on a Closed/Revoked // session MUST be rejected (A-523 session state machine). return nil, fmt.Errorf("bearers: session %q is terminal (%s), rejects frame", msg.SessionID, session.Status) } if session.IsExpired(nowUnix(sdkCtx)) { // TTL expiry transitions the session to Closed (the handler // enforces expiry on Send/Receive checks). session.Status = types.SessionClosed session.ClosedAt = nowUnix(sdkCtx) s.Keeper.SetSession(sdkCtx, session) return nil, fmt.Errorf("bearers: session %q expired (ttl %d), rejects frame", msg.SessionID, session.TTL) } frame := types.Frame{ FrameID: msg.FrameID, SenderReach: msg.Signer, PayloadBytes: msg.PayloadBytes, SentAt: nowUnix(sdkCtx), } session.Frames = append(session.Frames, frame) s.Keeper.SetSession(sdkCtx, session) sdkCtx.EventManager().EmitEvent(sdk.NewEvent( "bearers.frame_sent", sdk.NewAttribute("session_id", msg.SessionID), sdk.NewAttribute("frame_id", msg.FrameID), sdk.NewAttribute("sender_reach", msg.Signer), // NO geolocation (A-522 surveillance-resistant invariant). )) return &types.MsgSendOYSATFrameResponse{}, nil } // --- ReceiveOYSATFrame (ack a frame; Open → Active on first ack) ------------- // ReceiveOYSATFrame acknowledges receipt of an OY-SAT frame. The handler // transitions the session Open → Active on the first ack. The handler // enforces the stateful session-status check: the session must be Open or // Active (acks on Closed/Revoked are REJECTED — the rejected-frame case). func (s msgServer) ReceiveOYSATFrame(ctx interface{}, msg *types.MsgReceiveOYSATFrame) (*types.MsgReceiveOYSATFrameResponse, error) { if err := msg.ValidateBasic(); err != nil { return nil, err } sdkCtx := unwrapCtx(ctx) session, ok := s.Keeper.GetSession(sdkCtx, msg.SessionID) if !ok { return nil, fmt.Errorf("bearers: session %q not found", msg.SessionID) } if session.IsTerminal() { // Rejected-frame case: an ack received on a Closed/Revoked // session MUST be rejected (A-523 session state machine). return nil, fmt.Errorf("bearers: session %q is terminal (%s), rejects ack", msg.SessionID, session.Status) } if session.IsExpired(nowUnix(sdkCtx)) { session.Status = types.SessionClosed session.ClosedAt = nowUnix(sdkCtx) s.Keeper.SetSession(sdkCtx, session) return nil, fmt.Errorf("bearers: session %q expired (ttl %d), rejects ack", msg.SessionID, session.TTL) } // Find the named frame; mark it received. found := false for i := range session.Frames { if session.Frames[i].FrameID == msg.FrameID { session.Frames[i].Received = true found = true break } } if !found { return nil, fmt.Errorf("bearers: frame %q not found on session %q", msg.FrameID, msg.SessionID) } // Open → Active on the first ack. if session.Status == types.SessionOpen { session.Status = types.SessionActive } s.Keeper.SetSession(sdkCtx, session) sdkCtx.EventManager().EmitEvent(sdk.NewEvent( "bearers.frame_received", sdk.NewAttribute("session_id", msg.SessionID), sdk.NewAttribute("frame_id", msg.FrameID), sdk.NewAttribute("status", string(session.Status)), // NO geolocation (A-522 surveillance-resistant invariant). )) return &types.MsgReceiveOYSATFrameResponse{}, nil } // --- IssueOYQR (issue a one-shot OY-QR, consumed=false) ---------------------- // IssueOYQR issues a one-shot OY-QR (consumed=false). The handler enforces // idempotency (qr-id must not already exist). func (s msgServer) IssueOYQR(ctx interface{}, msg *types.MsgIssueOYQR) (*types.MsgIssueOYQRResponse, error) { if err := msg.ValidateBasic(); err != nil { return nil, err } sdkCtx := unwrapCtx(ctx) // Idempotency: qr-id must not already exist. if _, ok := s.Keeper.GetOYQRCode(sdkCtx, msg.QRID); ok { return nil, fmt.Errorf("bearers: qr %q already exists", msg.QRID) } qr := types.OYQRCode{ QRID: msg.QRID, PayloadBytes: msg.PayloadBytes, Consumed: false, IssuerReachID: msg.IssuerReachID, AmountGrain: msg.AmountGrain, ExpiresAt: msg.ExpiresAt, } s.Keeper.SetOYQRCode(sdkCtx, qr) sdkCtx.EventManager().EmitEvent(sdk.NewEvent( "bearers.qr_issued", sdk.NewAttribute("qr_id", msg.QRID), sdk.NewAttribute("issuer_reach", msg.IssuerReachID), sdk.NewAttribute("amount_grain", fmt.Sprintf("%d", msg.AmountGrain)), sdk.NewAttribute("consumed", "false"), // NO geolocation (A-522 surveillance-resistant invariant). )) return &types.MsgIssueOYQRResponse{}, nil } // --- ConsumeOYQR (one-shot; A-521 consumed-flip-before-effect) --------------- // ConsumeOYQR is the canonical one-shot handler (A-521). The ordering is: // 1. load QR // 2. assert !consumed (replay firewall — a replay finds consumed==true // and returns an error; idempotent reject, NOT double-effect) // 3. assert expires-at > now (the QR is still valid) // 4. FLIP consumed=true (state write FIRST — A-521) // 5. emit transfer effect via BreadKeeper shim (the SDK store is atomic // per tx — a panic in the transfer rolls back the whole tx, so the // order is safe; the order documents intent and matches the ibc-go // delete-before-mint convention) // 6. emit event // 7. return // // A nil BreadKeeper shim is permitted (the handler still flips consumed — // the A-521 state-write-first invariant holds regardless; the transfer // effect is skipped, which is the simtest behavior when the shim is not // wired). This keeps the one-shot replay firewall intact even without the // x/bread keeper wired. func (s msgServer) ConsumeOYQR(ctx interface{}, msg *types.MsgConsumeOYQR) (*types.MsgConsumeOYQRResponse, error) { if err := msg.ValidateBasic(); err != nil { return nil, err } sdkCtx := unwrapCtx(ctx) // 1. Load QR. qr, ok := s.Keeper.GetOYQRCode(sdkCtx, msg.QRID) if !ok { return nil, fmt.Errorf("bearers: qr %q not found", msg.QRID) } // 2. Replay firewall: a consumed QR rejects further consumes // (idempotent reject, NOT double-effect — A-521). if qr.Consumed { return nil, fmt.Errorf("bearers: qr %q already consumed (one-shot — A-521)", msg.QRID) } // 3. Expiry check: the QR must still be valid (expires-at > now). now := nowUnix(sdkCtx) if qr.ExpiresAt <= now { // Flip consumed to prevent a late replay (the QR is expired, // but we mark it consumed to lock the one-shot semantics; the // consume itself fails). qr.Consumed = true s.Keeper.SetOYQRCode(sdkCtx, qr) return nil, fmt.Errorf("bearers: qr %q expired (expires-at %d <= now %d)", msg.QRID, qr.ExpiresAt, now) } // 4. FLIP consumed=true (state write FIRST — A-521). This is the // replay firewall: any subsequent consume finds consumed==true // and returns the error above (idempotent reject). qr.Consumed = true s.Keeper.SetOYQRCode(sdkCtx, qr) // 5. Emit transfer effect via BreadKeeper shim. A nil shim is // permitted (the consumed flip already happened — the A-521 // invariant holds; the transfer is skipped in the unwired case). var transferErr error if s.Keeper.breadKeeper != nil { transferErr = s.Keeper.breadKeeper.TransferGrain(qr.IssuerReachID, msg.ConsumerReachID, qr.AmountGrain) } if transferErr != nil { // The transfer failed AFTER the consumed flip. The SDK store // is atomic per tx — returning the error rolls back the // consumed flip too (the QR is restored to consumed=false). // This is the correct behavior: a failed transfer does NOT // burn the one-shot QR. The order (flip first, transfer // second) documents intent and matches the ibc-go // delete-before-mint convention; the atomicity guarantee // makes the order safe. return nil, fmt.Errorf("bearers: qr %q transfer effect failed: %w", msg.QRID, transferErr) } // 6. Emit event. sdkCtx.EventManager().EmitEvent(sdk.NewEvent( "bearers.qr_consumed", sdk.NewAttribute("qr_id", msg.QRID), sdk.NewAttribute("issuer_reach", qr.IssuerReachID), sdk.NewAttribute("consumer_reach", msg.ConsumerReachID), sdk.NewAttribute("amount_grain", fmt.Sprintf("%d", qr.AmountGrain)), sdk.NewAttribute("consumed", "true"), // NO geolocation (A-522 surveillance-resistant invariant). )) return &types.MsgConsumeOYQRResponse{}, nil }