11package proxy
22
33import (
4+ "encoding/json"
5+ "fmt"
6+ "net/http"
7+
48 "github.com/gin-gonic/gin"
9+ "github.com/mitchellh/mapstructure"
510 "github.com/prometheus/client_golang/prometheus"
6- "github.com/scroll-tech/go-ethereum/params"
7- "gorm.io/gorm"
11+ "github.com/scroll-tech/go-ethereum/log"
812
9- "scroll-tech/common/types/message "
13+ "scroll-tech/common/types"
1014
1115 "scroll-tech/coordinator/internal/config"
12- "scroll-tech/coordinator/internal/logic/provertask"
13- "scroll-tech/coordinator/internal/logic/verifier"
1416 coordinatorType "scroll-tech/coordinator/internal/types"
1517)
1618
19+ func getSessionData (ctx * gin.Context ) (string , * coordinatorType.LoginParameter ) {
20+
21+ publicKeyData , publicKeyExist := ctx .Get (coordinatorType .PublicKey )
22+ publicKey , castOk := publicKeyData .(string )
23+ if ! publicKeyExist || ! castOk {
24+ nerr := fmt .Errorf ("no public key binding: %v" , publicKeyData )
25+ log .Warn ("get_task parameter fail" , "error" , nerr )
26+
27+ types .RenderFailure (ctx , types .ErrCoordinatorParameterInvalidNo , nerr )
28+ return "" , nil
29+ }
30+
31+ loginParamData , publicKeyExist := ctx .Get (LoginParamCache )
32+ loginParam , castOk := loginParamData .(* coordinatorType.LoginParameter )
33+ if ! publicKeyExist || ! castOk {
34+ nerr := fmt .Errorf ("no login param binding: %v" , loginParamData )
35+ log .Warn ("get_task parameter fail" , "error" , nerr )
36+
37+ types .RenderFailure (ctx , types .ErrCoordinatorParameterInvalidNo , nerr )
38+ return "" , nil
39+ }
40+
41+ return publicKey , loginParam
42+ }
43+
1744// GetTaskController the get prover task api controller
1845type GetTaskController struct {
19- proverTasks map [message.ProofType ]provertask.ProverTask
46+ tokenCache * UserTokenCache
47+ clients Clients
48+ priorityUpstream map [string ]string
2049
2150 getTaskAccessCounter * prometheus.CounterVec
2251}
2352
2453// NewGetTaskController create a get prover task controller
25- func NewGetTaskController (cfg * config.Config , chainCfg * params. ChainConfig , db * gorm. DB , verifier * verifier. Verifier , reg prometheus.Registerer ) * GetTaskController {
54+ func NewGetTaskController (cfg * config.Config , clients Clients , tokenCache * UserTokenCache , reg prometheus.Registerer ) * GetTaskController {
2655 // TODO: implement proxy get task controller initialization
2756 return & GetTaskController {
28- proverTasks : make (map [message.ProofType ]provertask.ProverTask ),
57+ priorityUpstream : make (map [string ]string ),
58+ tokenCache : tokenCache ,
59+ clients : clients ,
2960 }
3061}
3162
@@ -36,10 +67,95 @@ func (ptc *GetTaskController) incGetTaskAccessCounter(ctx *gin.Context) error {
3667
3768// GetTasks get assigned chunk/batch task
3869func (ptc * GetTaskController ) GetTasks (ctx * gin.Context ) {
39- // TODO: implement proxy get tasks logic
70+ var getTaskParameter coordinatorType.GetTaskParameter
71+ if err := ctx .ShouldBind (& getTaskParameter ); err != nil {
72+ nerr := fmt .Errorf ("prover task parameter invalid, err:%w" , err )
73+ types .RenderFailure (ctx , types .ErrCoordinatorParameterInvalidNo , nerr )
74+ return
75+ }
76+
77+ publicKey , loginParam := getSessionData (ctx )
78+ if publicKey == "" || loginParam == nil {
79+ return
80+ }
81+
82+ tokens := ptc .tokenCache .Get (publicKey )
83+
84+ onClientFail := func (upstream string ) {
85+ //TODO: log re-connect request in info level
86+
87+ request := TokenUpdate {
88+ PublicKey : publicKey ,
89+ Upstream : upstream ,
90+ Phase : tokens .LoginPhase ,
91+ LoginParam : * loginParam ,
92+ CompleteNotify : nil ,
93+ }
94+ select {
95+ case <- ctx .Done ():
96+ case ptc .tokenCache .tokenCacheUpdate <- & request :
97+ }
98+
99+ }
100+
101+ priorityUpstream , exist := ptc .priorityUpstream [publicKey ]
102+ if exist {
103+ cli := ptc .clients [priorityUpstream ]
104+ loginSchema := tokens .LoginData [priorityUpstream ]
105+ if loginSchema == nil {
106+ onClientFail (priorityUpstream )
107+ } else {
108+ ret , triggerUpdate := getTaskFromClient (ctx , cli , & getTaskParameter , loginSchema .Token )
109+ if ret != nil {
110+
111+ } else if triggerUpdate {
112+ onClientFail (priorityUpstream )
113+ }
114+ }
115+ types .RenderFailure (ctx , types .ErrCoordinatorEmptyProofData , fmt .Errorf ("get empty prover task" ))
116+ }
117+
118+ for n , cli := range ptc .clients {
119+
120+ }
40121}
41122
42- func (ptc * GetTaskController ) proofType (para * coordinatorType.GetTaskParameter ) message.ProofType {
43- // TODO: implement proxy proof type logic
44- return message .ProofTypeChunk
45- }
123+ func getTaskFromClient (ctx * gin.Context , cli Client , param * coordinatorType.GetTaskParameter , token string ) (* coordinatorType.GetTaskSchema , bool ) {
124+
125+ theCli := cli .PeekClient ()
126+ if theCli == nil {
127+ return nil , true
128+ }
129+
130+ resp , err := theCli .GetTask (ctx , param , token )
131+ if err != nil {
132+ // log the err in error level
133+ return nil , false
134+ }
135+
136+ // Parse response
137+ if resp .StatusCode == http .StatusOK || resp .StatusCode == http .StatusUnauthorized {
138+ unAuth := resp .StatusCode == http .StatusUnauthorized
139+ var respWithData types.Response
140+ // Note: Body is consumed after decoding, caller should not read it again
141+ if err := json .NewDecoder (resp .Body ).Decode (& respWithData ); err == nil {
142+ if unAuth && respWithData .ErrCode == types .ErrJWTTokenExpired {
143+ return nil , true
144+ }
145+
146+ var getTaskResult coordinatorType.GetTaskSchema
147+ err = mapstructure .Decode (respWithData .Data , & getTaskResult )
148+ if err != nil {
149+ log .Error ("parse get task data fail" , "respdata" , respWithData .Data )
150+ return nil , false
151+ }
152+ return & getTaskResult , false
153+ } else {
154+ log .Error ("parse get task response failed" , "error" , err )
155+ //fmt.Errorf("login parsing response failed: %v", err)
156+ return nil , false
157+ }
158+ }
159+
160+ return nil , false
161+ }
0 commit comments