-
Notifications
You must be signed in to change notification settings - Fork 42
/
keeper.go
130 lines (107 loc) · 3.44 KB
/
keeper.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
package keeper
import (
"context"
"github.com/omni-network/omni/halo/portal/types"
"github.com/omni-network/omni/lib/errors"
"github.com/omni-network/omni/lib/xchain"
ormv1alpha1 "cosmossdk.io/api/cosmos/orm/v1alpha1"
"cosmossdk.io/core/store"
"cosmossdk.io/orm/model/ormdb"
"cosmossdk.io/orm/types/ormerrors"
sdk "github.com/cosmos/cosmos-sdk/types"
)
type Keeper struct {
blockTable BlockTable
msgTable MsgTable
offsetTable OffsetTable
}
func NewKeeper(storeService store.KVStoreService) (Keeper, error) {
schema := &ormv1alpha1.ModuleSchemaDescriptor{SchemaFile: []*ormv1alpha1.ModuleSchemaDescriptor_FileEntry{
{Id: 1, ProtoFileName: File_halo_portal_keeper_portal_proto.Path()},
}}
modDB, err := ormdb.NewModuleDB(schema, ormdb.ModuleDBOptions{KVStoreService: storeService})
if err != nil {
return Keeper{}, errors.Wrap(err, "create module db")
}
portalStore, err := NewPortalStore(modDB)
if err != nil {
return Keeper{}, errors.Wrap(err, "create portal store")
}
return Keeper{
blockTable: portalStore.BlockTable(),
msgTable: portalStore.MsgTable(),
offsetTable: portalStore.OffsetTable(),
}, nil
}
func (k Keeper) EmitMsg(ctx sdk.Context, typ types.MsgType, msgTypeID uint64, destChainID uint64, shardID xchain.ShardID) (uint64, error) {
if (destChainID == xchain.BroadcastChainID) != shardID.Broadcast() {
return 0, errors.New("dest chain and shard broadcast flag mismatch [BUG]")
}
height := uint64(ctx.BlockHeight())
// Get or create a block to add the message to
var blockID uint64
if block, err := k.blockTable.GetByCreatedHeight(ctx, height); ormerrors.IsNotFound(err) {
blockID, err = k.blockTable.InsertReturningId(ctx, &Block{CreatedHeight: height})
if err != nil {
return 0, errors.Wrap(err, "insert block")
}
} else if err != nil {
return 0, errors.Wrap(err, "get block")
} else {
blockID = block.GetId()
}
offset, err := k.incAndGetOffset(ctx, destChainID, shardID)
if err != nil {
return 0, errors.Wrap(err, "increment offset")
}
err = k.msgTable.Insert(ctx, &Msg{
BlockId: blockID,
MsgType: uint32(typ),
MsgTypeId: msgTypeID,
DestChainId: destChainID,
ShardId: uint64(shardID),
StreamOffset: offset,
})
if err != nil {
return 0, errors.Wrap(err, "insert message")
}
return blockID, nil
}
func (k Keeper) incAndGetOffset(ctx context.Context, destChainID uint64, shardID xchain.ShardID) (uint64, error) {
offset, err := k.offsetTable.GetByDestChainIdShardId(ctx, destChainID, uint64(shardID))
if ormerrors.IsNotFound(err) {
offset = &Offset{
DestChainId: destChainID,
ShardId: uint64(shardID),
Offset: 0,
}
} else if err != nil {
return 0, errors.Wrap(err, "get next offset")
}
// Increment the offset
offset.Offset++
// Save the new offset
if err := k.offsetTable.Save(ctx, offset); err != nil {
return 0, errors.Wrap(err, "save next offset")
}
return offset.GetOffset(), nil
}
func (k Keeper) getBlockAndMsgs(ctx context.Context, blockID uint64) (*Block, []*Msg, error) {
block, err := k.blockTable.Get(ctx, blockID)
if err != nil {
return nil, nil, errors.Wrap(err, "get block")
}
iter, err := k.msgTable.List(ctx, MsgBlockIdIndexKey{}.WithBlockId(blockID))
if err != nil {
return nil, nil, errors.Wrap(err, "list messages")
}
var msgs []*Msg
for iter.Next() {
msg, err := iter.Value()
if err != nil {
return nil, nil, errors.Wrap(err, "get msg value")
}
msgs = append(msgs, msg)
}
return block, msgs, nil
}