-
Notifications
You must be signed in to change notification settings - Fork 2
chore: enable SDK functionality for primitive stream #164
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Merged
Merged
Changes from all commits
Commits
File filter
Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,224 @@ | ||
| package tnclient | ||
|
|
||
| import ( | ||
| "context" | ||
| "fmt" | ||
| "strconv" | ||
|
|
||
| "github.com/cockroachdb/apd/v3" | ||
| "github.com/pkg/errors" | ||
| kwilClientType "github.com/trufnetwork/kwil-db/core/client/types" | ||
| kwilType "github.com/trufnetwork/kwil-db/core/types" | ||
| "github.com/trufnetwork/kwil-db/node/types" | ||
| tn_api "github.com/trufnetwork/sdk-go/core/contractsapi" | ||
| clientType "github.com/trufnetwork/sdk-go/core/types" | ||
| "github.com/trufnetwork/sdk-go/core/util" | ||
| ) | ||
|
|
||
| // TransportAction implements IAction interface using the Transport abstraction. | ||
| // This allows actions to work with any transport (HTTP, CRE, etc.). | ||
| // | ||
| // MINIMAL IMPLEMENTATION: Only GetRecord is fully implemented. | ||
| // Other methods return "not implemented" errors since they're not needed by QuantAMM. | ||
| type TransportAction struct { | ||
| transport Transport | ||
| } | ||
|
|
||
| var _ clientType.IAction = (*TransportAction)(nil) | ||
|
|
||
| // GetRecord reads the records of the stream within the given date range. | ||
| // This is a core method needed by QuantAMM for reading stream data. | ||
| func (a *TransportAction) GetRecord(ctx context.Context, input clientType.GetRecordInput) (clientType.ActionResult, error) { | ||
| var args []any | ||
| args = append(args, input.DataProvider) | ||
| args = append(args, input.StreamId) | ||
| args = append(args, util.TransformOrNil(input.From, func(date int) any { return date })) | ||
| args = append(args, util.TransformOrNil(input.To, func(date int) any { return date })) | ||
| args = append(args, util.TransformOrNil(input.FrozenAt, func(date int) any { return date })) | ||
| args = append(args, util.TransformOrNil(input.BaseDate, func(date int) any { return date })) | ||
| if input.UseCache != nil { | ||
| args = append(args, *input.UseCache) | ||
| } | ||
|
|
||
| prefix := "" | ||
| if input.Prefix != nil { | ||
| prefix = *input.Prefix | ||
| } | ||
|
|
||
| result, err := a.transport.Call(ctx, "", prefix+"get_record", args) | ||
| if err != nil { | ||
| return clientType.ActionResult{}, errors.WithStack(err) | ||
| } | ||
|
|
||
| // Decode raw SQL output using shared type | ||
| rawOutputs, err := tn_api.DecodeCallResult[clientType.GetRecordRawOutput](result.QueryResult) | ||
| if err != nil { | ||
| return clientType.ActionResult{}, errors.WithStack(err) | ||
| } | ||
|
|
||
| // Parse strings to proper types | ||
| var outputs []clientType.StreamResult | ||
| for _, rawOutput := range rawOutputs { | ||
| value, _, err := apd.NewFromString(rawOutput.Value) | ||
| if err != nil { | ||
| return clientType.ActionResult{}, errors.WithStack(err) | ||
| } | ||
|
|
||
| eventTime := 0 | ||
| if rawOutput.EventTime != "" { | ||
| eventTime, err = strconv.Atoi(rawOutput.EventTime) | ||
| if err != nil { | ||
| return clientType.ActionResult{}, errors.WithStack(err) | ||
| } | ||
| } | ||
|
|
||
| outputs = append(outputs, clientType.StreamResult{ | ||
| EventTime: eventTime, | ||
| Value: *value, | ||
| }) | ||
| } | ||
|
|
||
| // Note: Cache metadata parsing is not implemented in this minimal version | ||
| // as noted in TRANSPORT_IMPLEMENTATION_NOTES.md | ||
| return clientType.ActionResult{Results: outputs}, nil | ||
| } | ||
|
|
||
| // Stub implementations for IAction methods not needed by QuantAMM. | ||
| // These return errors indicating they're not implemented for custom transports. | ||
|
|
||
| func (a *TransportAction) ExecuteProcedure(ctx context.Context, procedure string, args [][]any) (types.Hash, error) { | ||
| return a.transport.Execute(ctx, "", procedure, args) | ||
| } | ||
|
|
||
| func (a *TransportAction) CallProcedure(ctx context.Context, procedure string, args []any) (*kwilType.QueryResult, error) { | ||
| result, err := a.transport.Call(ctx, "", procedure, args) | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
| return result.QueryResult, nil | ||
| } | ||
|
|
||
| func (a *TransportAction) GetIndex(ctx context.Context, input clientType.GetIndexInput) (clientType.ActionResult, error) { | ||
| return clientType.ActionResult{}, fmt.Errorf("GetIndex not implemented for custom transports - use HTTP transport or implement if needed") | ||
| } | ||
|
|
||
| func (a *TransportAction) GetIndexChange(ctx context.Context, input clientType.GetIndexChangeInput) (clientType.ActionResult, error) { | ||
| return clientType.ActionResult{}, fmt.Errorf("GetIndexChange not implemented for custom transports - use HTTP transport or implement if needed") | ||
| } | ||
|
|
||
| func (a *TransportAction) GetType(ctx context.Context, locator clientType.StreamLocator) (clientType.StreamType, error) { | ||
| return "", fmt.Errorf("GetType not implemented for custom transports - use HTTP transport or implement if needed") | ||
| } | ||
|
|
||
| func (a *TransportAction) GetFirstRecord(ctx context.Context, input clientType.GetFirstRecordInput) (clientType.ActionResult, error) { | ||
| return clientType.ActionResult{}, fmt.Errorf("GetFirstRecord not implemented for custom transports - use HTTP transport or implement if needed") | ||
| } | ||
|
|
||
| func (a *TransportAction) SetReadVisibility(ctx context.Context, input clientType.VisibilityInput) (types.Hash, error) { | ||
| return types.Hash{}, fmt.Errorf("SetReadVisibility not implemented for custom transports - use HTTP transport or implement if needed") | ||
| } | ||
|
|
||
| func (a *TransportAction) GetReadVisibility(ctx context.Context, locator clientType.StreamLocator) (*util.VisibilityEnum, error) { | ||
| return nil, fmt.Errorf("GetReadVisibility not implemented for custom transports - use HTTP transport or implement if needed") | ||
| } | ||
|
|
||
| func (a *TransportAction) SetComposeVisibility(ctx context.Context, input clientType.VisibilityInput) (types.Hash, error) { | ||
| return types.Hash{}, fmt.Errorf("SetComposeVisibility not implemented for custom transports - use HTTP transport or implement if needed") | ||
| } | ||
|
|
||
| func (a *TransportAction) GetComposeVisibility(ctx context.Context, locator clientType.StreamLocator) (*util.VisibilityEnum, error) { | ||
| return nil, fmt.Errorf("GetComposeVisibility not implemented for custom transports - use HTTP transport or implement if needed") | ||
| } | ||
|
|
||
| func (a *TransportAction) AllowReadWallet(ctx context.Context, input clientType.ReadWalletInput) (types.Hash, error) { | ||
| return types.Hash{}, fmt.Errorf("AllowReadWallet not implemented for custom transports - use HTTP transport or implement if needed") | ||
| } | ||
|
|
||
| func (a *TransportAction) DisableReadWallet(ctx context.Context, input clientType.ReadWalletInput) (types.Hash, error) { | ||
| return types.Hash{}, fmt.Errorf("DisableReadWallet not implemented for custom transports - use HTTP transport or implement if needed") | ||
| } | ||
|
|
||
| func (a *TransportAction) AllowComposeStream(ctx context.Context, locator clientType.StreamLocator) (types.Hash, error) { | ||
| return types.Hash{}, fmt.Errorf("AllowComposeStream not implemented for custom transports - use HTTP transport or implement if needed") | ||
| } | ||
|
|
||
| func (a *TransportAction) DisableComposeStream(ctx context.Context, locator clientType.StreamLocator) (types.Hash, error) { | ||
| return types.Hash{}, fmt.Errorf("DisableComposeStream not implemented for custom transports - use HTTP transport or implement if needed") | ||
| } | ||
|
|
||
| func (a *TransportAction) GetStreamOwner(ctx context.Context, locator clientType.StreamLocator) ([]byte, error) { | ||
| return nil, fmt.Errorf("GetStreamOwner not implemented for custom transports - use HTTP transport or implement if needed") | ||
| } | ||
|
|
||
| func (a *TransportAction) GetAllowedReadWallets(ctx context.Context, locator clientType.StreamLocator) ([]util.EthereumAddress, error) { | ||
| return nil, fmt.Errorf("GetAllowedReadWallets not implemented for custom transports - use HTTP transport or implement if needed") | ||
| } | ||
|
|
||
| func (a *TransportAction) GetAllowedComposeStreams(ctx context.Context, locator clientType.StreamLocator) ([]clientType.StreamLocator, error) { | ||
| return nil, fmt.Errorf("GetAllowedComposeStreams not implemented for custom transports - use HTTP transport or implement if needed") | ||
| } | ||
|
|
||
| func (a *TransportAction) SetDefaultBaseTime(ctx context.Context, input clientType.DefaultBaseTimeInput) (types.Hash, error) { | ||
| return types.Hash{}, fmt.Errorf("SetDefaultBaseTime not implemented for custom transports - use HTTP transport or implement if needed") | ||
| } | ||
|
|
||
| func (a *TransportAction) BatchStreamExists(ctx context.Context, streams []clientType.StreamLocator) ([]clientType.StreamExistsResult, error) { | ||
| return nil, fmt.Errorf("BatchStreamExists not implemented for custom transports - use HTTP transport or implement if needed") | ||
| } | ||
|
|
||
| func (a *TransportAction) BatchFilterStreamsByExistence(ctx context.Context, streams []clientType.StreamLocator, returnExisting bool) ([]clientType.StreamLocator, error) { | ||
| return nil, fmt.Errorf("BatchFilterStreamsByExistence not implemented for custom transports - use HTTP transport or implement if needed") | ||
| } | ||
|
|
||
| // TransportPrimitiveAction implements IPrimitiveAction interface using the Transport abstraction. | ||
| // This allows primitive stream actions to work with any transport (HTTP, CRE, etc.). | ||
| // | ||
| // MINIMAL IMPLEMENTATION: Only InsertRecords is fully implemented. | ||
| // Other methods return "not implemented" errors since they're not needed by QuantAMM. | ||
| type TransportPrimitiveAction struct { | ||
| TransportAction | ||
| } | ||
|
|
||
| var _ clientType.IPrimitiveAction = (*TransportPrimitiveAction)(nil) | ||
|
|
||
| // InsertRecords inserts multiple records into primitive streams. | ||
| // This is a core method needed by QuantAMM for writing stream data. | ||
| func (p *TransportPrimitiveAction) InsertRecords(ctx context.Context, inputs []clientType.InsertRecordInput, opts ...kwilClientType.TxOpt) (types.Hash, error) { | ||
| var ( | ||
| dataProviders []string | ||
| streamIds []string | ||
| eventTimes []int | ||
| values kwilType.DecimalArray | ||
| ) | ||
|
|
||
| for _, input := range inputs { | ||
| // Convert float64 to decimal with 36 precision and 18 scale | ||
| valueNumeric, err := kwilType.ParseDecimalExplicit(strconv.FormatFloat(input.Value, 'f', -1, 64), 36, 18) | ||
| if err != nil { | ||
| return types.Hash{}, errors.WithStack(err) | ||
| } | ||
|
|
||
| dataProviders = append(dataProviders, input.DataProvider) | ||
| streamIds = append(streamIds, input.StreamId) | ||
| eventTimes = append(eventTimes, input.EventTime) | ||
| values = append(values, valueNumeric) | ||
| } | ||
|
|
||
| return p.transport.Execute(ctx, "", "insert_records", [][]any{{ | ||
| dataProviders, | ||
| streamIds, | ||
| eventTimes, | ||
| values, | ||
| }}, opts...) | ||
| } | ||
|
|
||
| // InsertRecord inserts a single record - stub implementation | ||
| func (p *TransportPrimitiveAction) InsertRecord(ctx context.Context, input clientType.InsertRecordInput, opts ...kwilClientType.TxOpt) (types.Hash, error) { | ||
| // Just delegate to InsertRecords | ||
| return p.InsertRecords(ctx, []clientType.InsertRecordInput{input}, opts...) | ||
| } | ||
|
|
||
| // CheckValidPrimitiveStream checks if the stream is a valid primitive stream - stub implementation | ||
| func (p *TransportPrimitiveAction) CheckValidPrimitiveStream(ctx context.Context, locator clientType.StreamLocator) error { | ||
| return fmt.Errorf("CheckValidPrimitiveStream not implemented for custom transports - use HTTP transport or implement if needed") | ||
| } | ||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,12 @@ | ||
| package types | ||
|
|
||
| // GetRecordRawOutput represents the raw SQL output from get_record, get_index, | ||
| // get_index_change, and get_first_record procedures. | ||
| // | ||
| // This struct is shared across multiple packages to avoid duplication: | ||
| // - core/contractsapi (HTTP-based implementations) | ||
| // - core/tnclient (transport-aware implementations) | ||
| type GetRecordRawOutput struct { | ||
| EventTime string `json:"event_time"` | ||
| Value string `json:"value"` | ||
| } |
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Uh oh!
There was an error while loading. Please reload this page.