@@ -2,22 +2,17 @@ package main
22
33import (
44 "context"
5- "github.com/data-preservation-programs/RetrievalBot/pkg/convert "
5+ "github.com/data-preservation-programs/RetrievalBot/integration/filplus/util "
66 "github.com/data-preservation-programs/RetrievalBot/pkg/env"
77 "github.com/data-preservation-programs/RetrievalBot/pkg/model"
8- "github.com/data-preservation-programs/RetrievalBot/pkg/requesterror"
98 "github.com/data-preservation-programs/RetrievalBot/pkg/resolver"
109 "github.com/data-preservation-programs/RetrievalBot/pkg/task"
11- "github.com/ipfs/go-cid"
1210 logging "github.com/ipfs/go-log/v2"
1311 _ "github.com/joho/godotenv/autoload"
14- "github.com/libp2p/go-libp2p/core/peer"
1512 "github.com/pkg/errors"
1613 "go.mongodb.org/mongo-driver/bson"
1714 "go.mongodb.org/mongo-driver/mongo"
1815 "go.mongodb.org/mongo-driver/mongo/options"
19- "golang.org/x/exp/slices"
20- "strconv"
2116 "time"
2217)
2318
@@ -87,7 +82,7 @@ func NewFilPlusIntegration() *FilPlusIntegration {
8782 }
8883
8984 // Check public IP address
90- ipInfo , err := resolver .GetPublicIPInfo (context . TODO () , "" , "" )
85+ ipInfo , err := resolver .GetPublicIPInfo (ctx , "" , "" )
9186 if err != nil {
9287 panic (err )
9388 }
@@ -107,76 +102,6 @@ func NewFilPlusIntegration() *FilPlusIntegration {
107102 }
108103}
109104
110- var moduleMetadataMap = map [task.ModuleName ]map [string ]string {
111- task .GraphSync : {
112- "assume_label" : "true" ,
113- "retrieve_type" : "root_block" ,
114- },
115- task .Bitswap : {
116- "assume_label" : "true" ,
117- "retrieve_type" : "root_block" ,
118- },
119- task .HTTP : {
120- "retrieve_type" : "piece" ,
121- "retrieve_size" : "1048576" ,
122- },
123- }
124-
125- func (f * FilPlusIntegration ) addErrorResults (results []interface {}, document model.DealState ,
126- providerInfo resolver.MinerInfo , location resolver.IPInfo ,
127- errorCode task.ErrorCode , errorMessage string ) []interface {} {
128- for module , metadata := range moduleMetadataMap {
129- newMetadata := make (map [string ]string )
130- for k , v := range metadata {
131- newMetadata [k ] = v
132- }
133- newMetadata ["deal_id" ] = strconv .Itoa (int (document .DealID ))
134- newMetadata ["client" ] = document .Client
135- results = append (results , task.Result {
136- Task : task.Task {
137- Requester : f .requester ,
138- Module : module ,
139- Metadata : newMetadata ,
140- Provider : task.Provider {
141- ID : document .Provider ,
142- PeerID : providerInfo .PeerId ,
143- Multiaddrs : convert .MultiaddrsBytesToStringArraySkippingError (providerInfo .Multiaddrs ),
144- City : location .City ,
145- Region : location .Region ,
146- Country : location .Country ,
147- Continent : location .Continent ,
148- },
149- Content : task.Content {
150- CID : document .Label ,
151- },
152- CreatedAt : time .Now ().UTC (),
153- Timeout : env .GetDuration (env .FilplusIntegrationTaskTimeout , 15 * time .Second )},
154- Retriever : task.Retriever {
155- PublicIP : f .ipInfo .IP ,
156- City : f .ipInfo .City ,
157- Region : f .ipInfo .Region ,
158- Country : f .ipInfo .Country ,
159- Continent : f .ipInfo .Continent ,
160- ASN : f .ipInfo .ASN ,
161- ISP : f .ipInfo .ISP ,
162- Latitude : f .ipInfo .Latitude ,
163- Longitude : f .ipInfo .Longitude ,
164- },
165- Result : task.RetrievalResult {
166- Success : false ,
167- ErrorCode : errorCode ,
168- ErrorMessage : errorMessage ,
169- TTFB : 0 ,
170- Speed : 0 ,
171- Duration : 0 ,
172- Downloaded : 0 ,
173- },
174- CreatedAt : time .Now ().UTC (),
175- })
176- }
177- return results
178- }
179-
180105func (f * FilPlusIntegration ) RunOnce (ctx context.Context ) error {
181106 logger .Info ("start running filplus integration" )
182107
@@ -220,112 +145,7 @@ func (f *FilPlusIntegration) RunOnce(ctx context.Context) error {
220145 }
221146
222147 documents = RandomObjects (documents , len (documents )/ 2 , f .randConst )
223- tasks := make ([]interface {}, 0 )
224- results := make ([]interface {}, 0 )
225- // Insert the documents into task queue
226- for _ , document := range documents {
227- // If the label is a correct CID, assume it is the payload CID and try GraphSync and Bitswap retrieval
228- labelCID , err := cid .Decode (document .Label )
229- if err != nil {
230- logger .With ("label" , document .Label , "deal_id" , document .DealID ).
231- Debug ("failed to decode label as CID" )
232- continue
233- }
234-
235- isPayloadCID := true
236- // Skip graphsync and bitswap if the cid is not decodable, i.e. it is a pieceCID
237- if ! slices .Contains ([]uint64 {cid .Raw , cid .DagCBOR , cid .DagProtobuf , cid .DagJSON , cid .DagJOSE },
238- labelCID .Prefix ().Codec ) {
239- logger .With ("provider" , document .Provider , "deal_id" , document .DealID ,
240- "label" , document .Label , "codec" , labelCID .Prefix ().Codec ).
241- Info ("Skip Bitswap and Graphsync because the Label is likely not a payload CID" )
242- isPayloadCID = false
243- }
244-
245- providerInfo , err := f .providerResolver .ResolveProvider (ctx , document .Provider )
246- if err != nil {
247- logger .With ("provider" , document .Provider , "deal_id" , document .DealID ).
248- Error ("failed to resolve provider" )
249- continue
250- }
251-
252- location , err := f .locationResolver .ResolveMultiaddrsBytes (ctx , providerInfo .Multiaddrs )
253- if err != nil {
254- if errors .As (err , & requesterror.BogonIPError {}) ||
255- errors .As (err , & requesterror.InvalidIPError {}) ||
256- errors .As (err , & requesterror.HostLookupError {}) ||
257- errors .As (err , & requesterror.NoValidMultiAddrError {}) {
258- results = f .addErrorResults (results , document , providerInfo , location ,
259- task .NoValidMultiAddrs , err .Error ())
260- } else {
261- logger .With ("provider" , document .Provider , "deal_id" , document .DealID , "err" , err ).
262- Error ("failed to resolve provider location" )
263- }
264- continue
265- }
266-
267- _ , err = peer .Decode (providerInfo .PeerId )
268- if err != nil {
269- logger .With ("provider" , document .Provider , "deal_id" , document .DealID , "peerID" , providerInfo .PeerId ,
270- "err" , err ).
271- Info ("failed to decode peerID" )
272- results = f .addErrorResults (results , document , providerInfo , location ,
273- task .InvalidPeerID , err .Error ())
274- continue
275- }
276-
277- if isPayloadCID {
278- for _ , module := range []task.ModuleName {task .GraphSync , task .Bitswap } {
279- tasks = append (tasks , task.Task {
280- Requester : f .requester ,
281- Module : module ,
282- Metadata : map [string ]string {
283- "deal_id" : strconv .Itoa (int (document .DealID )),
284- "client" : document .Client ,
285- "assume_label" : "true" ,
286- "retrieve_type" : "root_block" },
287- Provider : task.Provider {
288- ID : document .Provider ,
289- PeerID : providerInfo .PeerId ,
290- Multiaddrs : convert .MultiaddrsBytesToStringArraySkippingError (providerInfo .Multiaddrs ),
291- City : location .City ,
292- Region : location .Region ,
293- Country : location .Country ,
294- Continent : location .Continent ,
295- },
296- Content : task.Content {
297- CID : document .Label ,
298- },
299- CreatedAt : time .Now ().UTC (),
300- Timeout : env .GetDuration (env .FilplusIntegrationTaskTimeout , 15 * time .Second ),
301- })
302- }
303- }
304-
305- tasks = append (tasks , task.Task {
306- Requester : f .requester ,
307- Module : task .HTTP ,
308- Metadata : map [string ]string {
309- "deal_id" : strconv .Itoa (int (document .DealID )),
310- "client" : document .Client ,
311- "retrieve_type" : "piece" ,
312- "retrieve_size" : "1048576" },
313- Provider : task.Provider {
314- ID : document .Provider ,
315- PeerID : providerInfo .PeerId ,
316- Multiaddrs : convert .MultiaddrsBytesToStringArraySkippingError (providerInfo .Multiaddrs ),
317- City : location .City ,
318- Region : location .Region ,
319- Country : location .Country ,
320- Continent : location .Continent ,
321- },
322- Content : task.Content {
323- CID : document .PieceCID ,
324- },
325- CreatedAt : time .Now ().UTC (),
326- Timeout : env .GetDuration (env .FilplusIntegrationTaskTimeout , 15 * time .Second ),
327- })
328- }
148+ tasks , results := util .AddTasks (ctx , f .requester , f .ipInfo , documents , f .locationResolver , f .providerResolver )
329149
330150 if len (tasks ) > 0 {
331151 _ , err = f .taskCollection .InsertMany (ctx , tasks )
0 commit comments