-
Notifications
You must be signed in to change notification settings - Fork 19
/
oracle_data.go
183 lines (154 loc) · 6.2 KB
/
oracle_data.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
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
// Copyright (C) 2023 Gobalsky Labs Limited
//
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as
// published by the Free Software Foundation, either version 3 of the
// License, or (at your option) any later version.
//
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program. If not, see <http://www.gnu.org/licenses/>.
package sqlstore
import (
"context"
"fmt"
"code.vegaprotocol.io/vega/datanode/entities"
"code.vegaprotocol.io/vega/datanode/metrics"
v2 "code.vegaprotocol.io/vega/protos/data-node/api/v2"
"github.com/georgysavva/scany/pgxscan"
)
type OracleData struct {
*ConnectionSource
}
const (
sqlOracleDataColumns = `signers, data, meta_data, broadcast_at, error, tx_hash, vega_time, seq_num`
oracleDataQuery = `SELECT od.*, aggregated.spec_ids as matched_spec_ids
FROM
oracle_data od
LEFT JOIN LATERAL (
SELECT ARRAY_AGG(spec_id) AS spec_ids
FROM oracle_data_oracle_specs ods
WHERE od.vega_time = ods.vega_time
AND od.seq_num = ods.seq_num
) aggregated ON true
`
)
var oracleDataOrdering = TableOrdering{
ColumnOrdering{Name: "vega_time", Sorting: ASC},
ColumnOrdering{Name: "signers", Sorting: ASC},
}
func NewOracleData(connectionSource *ConnectionSource) *OracleData {
return &OracleData{
ConnectionSource: connectionSource,
}
}
func (od *OracleData) Add(ctx context.Context, oracleData *entities.OracleData) error {
defer metrics.StartSQLQuery("OracleData", "Add")()
query := fmt.Sprintf("insert into oracle_data(%s) values ($1, $2, $3, $4, $5, $6, $7, $8)", sqlOracleDataColumns)
if _, err := od.Connection.Exec(
ctx, query,
oracleData.ExternalData.Data.Signers, oracleData.ExternalData.Data.Data, oracleData.ExternalData.Data.MetaData,
oracleData.ExternalData.Data.BroadcastAt,
oracleData.ExternalData.Data.Error, oracleData.ExternalData.Data.TxHash,
oracleData.ExternalData.Data.VegaTime, oracleData.ExternalData.Data.SeqNum,
); err != nil {
err = fmt.Errorf("could not insert oracle data into database: %w", err)
return err
}
query2 := "insert into oracle_data_oracle_specs(vega_time, seq_num, spec_id) values ($1, $2, unnest($3::bytea[]))"
if _, err := od.Connection.Exec(
ctx, query2,
oracleData.ExternalData.Data.VegaTime, oracleData.ExternalData.Data.SeqNum, oracleData.ExternalData.Data.MatchedSpecIds,
); err != nil {
err = fmt.Errorf("could not insert oracle data join into database: %w", err)
return err
}
return nil
}
func (od *OracleData) ListOracleData(ctx context.Context, id string, pagination entities.Pagination) ([]entities.OracleData, entities.PageInfo, error) {
switch p := pagination.(type) {
case entities.CursorPagination:
return listOracleDataBySpecIDCursorPagination(ctx, od.Connection, id, p)
default:
panic("unsupported pagination")
}
}
func (od *OracleData) GetByTxHash(ctx context.Context, txHash entities.TxHash) ([]entities.OracleData, error) {
defer metrics.StartSQLQuery("OracleData", "GetByTxHash")()
var data []entities.Data
query := fmt.Sprintf(`%s WHERE tx_hash = $1`, oracleDataQuery)
err := pgxscan.Select(ctx, od.Connection, &data, query, txHash)
if err != nil {
return nil, err
}
return scannedDataToOracleData(data), nil
}
func scannedDataToOracleData(scanned []entities.Data) []entities.OracleData {
oracleData := []entities.OracleData{}
if len(scanned) > 0 {
for _, s := range scanned {
oracleData = append(oracleData, entities.OracleData{
ExternalData: &entities.ExternalData{
Data: &entities.Data{
Signers: s.Signers,
Data: s.Data,
MetaData: s.MetaData,
MatchedSpecIds: s.MatchedSpecIds,
BroadcastAt: s.BroadcastAt,
Error: s.Error,
TxHash: s.TxHash,
VegaTime: s.VegaTime,
SeqNum: s.SeqNum,
},
},
})
}
}
return oracleData
}
func listOracleDataBySpecIDCursorPagination(ctx context.Context, conn Connection, id string, pagination entities.CursorPagination) (
[]entities.OracleData, entities.PageInfo, error,
) {
var (
oracleData []entities.OracleData
data = []entities.Data{}
pageInfo entities.PageInfo
bindVars []interface{}
err error
)
query := oracleDataQuery
andOrWhere := "WHERE"
if len(id) > 0 {
specID := entities.SpecID(id)
query = fmt.Sprintf(`%s
WHERE EXISTS (SELECT 1 from oracle_data_oracle_specs ods
WHERE od.vega_time = ods.vega_time
AND od.seq_num = ods.seq_num
AND ods.spec_id=%s)`, query, nextBindVar(&bindVars, specID))
andOrWhere = "AND"
}
// if the cursor is empty, we should restrict the query to the last day of data as otherwise, the query will scan the full hypertable
// we only do this if we are returning the newest first data because that should be kept in memory by TimescaleDB anyway.
// If we have a first N cursor traversing newest first data, without an after cursor, we should also restrict by date.
// Traversing from the oldest data to the newest data will result in table scans and take time as we don't know what the oldest data is due to retention policies.
// Anything after the first page will have a vega time in the cursor so this will not be needed.
if pagination.HasForward() && !pagination.Forward.HasCursor() && pagination.NewestFirst {
query = fmt.Sprintf("%s %s vega_time > now() - interval '1 day'", query, andOrWhere)
}
query, bindVars, err = PaginateQuery[entities.OracleDataCursor](query, bindVars, oracleDataOrdering, pagination)
if err != nil {
return oracleData, pageInfo, err
}
defer metrics.StartSQLQuery("OracleData", "ListOracleData")()
// NOTE: If any error during the scan occurred, we return empty oracle data object.
if err = pgxscan.Select(ctx, conn, &data, query, bindVars...); err != nil {
return oracleData, pageInfo, err
}
oracleData = scannedDataToOracleData(data)
oracleData, pageInfo = entities.PageEntities[*v2.OracleDataEdge](oracleData, pagination)
return oracleData, pageInfo, nil
}