openyield/x/bearers/keeper/msg_server.go

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
}