From ad3bf689790a7a20753be96ecf0d87a3faa188db Mon Sep 17 00:00:00 2001 From: Lucas Jacques Date: Mon, 19 Aug 2024 12:38:42 +0200 Subject: [PATCH] feat: corroclient --- LICENSE | 7 ++ README.md | 5 ++ corroclient.go | 30 ++++++++ exec.go | 70 +++++++++++++++++ go.mod | 3 + queries.go | 104 ++++++++++++++++++++++++++ rows.go | 75 +++++++++++++++++++ scan.go | 74 ++++++++++++++++++ subscriptions.go | 191 +++++++++++++++++++++++++++++++++++++++++++++++ types.go | 141 ++++++++++++++++++++++++++++++++++ 10 files changed, 700 insertions(+) create mode 100644 LICENSE create mode 100644 README.md create mode 100644 corroclient.go create mode 100644 exec.go create mode 100644 go.mod create mode 100644 queries.go create mode 100644 rows.go create mode 100644 scan.go create mode 100644 subscriptions.go create mode 100644 types.go diff --git a/LICENSE b/LICENSE new file mode 100644 index 0000000..12e1202 --- /dev/null +++ b/LICENSE @@ -0,0 +1,7 @@ +Copyright 2024 SAS Valyent + +Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the “Software”), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions: + +The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software. + +THE SOFTWARE IS PROVIDED “AS IS”, WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. \ No newline at end of file diff --git a/README.md b/README.md new file mode 100644 index 0000000..309413b --- /dev/null +++ b/README.md @@ -0,0 +1,5 @@ +# corroclient +A go client for the [https://github.com/superfly/corrosion](Corrosion) API. + +# License +This project is licensed under the MIT License - see the [LICENSE](LICENSE) file for details. \ No newline at end of file diff --git a/corroclient.go b/corroclient.go new file mode 100644 index 0000000..3215045 --- /dev/null +++ b/corroclient.go @@ -0,0 +1,30 @@ +// This package is used to interact with the corrosion API. +package corroclient + +import "net/http" + +type Config struct { + URL string + Bearer string +} + +type CorroClient struct { + c *http.Client + url string + bearer string +} + +func (c *CorroClient) getURL(path string) string { + return c.url + path +} + +func NewCorroClient(config Config) *CorroClient { + client := &http.Client{} + corroClient := &CorroClient{ + c: client, + url: config.URL, + bearer: config.Bearer, + } + + return corroClient +} diff --git a/exec.go b/exec.go new file mode 100644 index 0000000..e296ad6 --- /dev/null +++ b/exec.go @@ -0,0 +1,70 @@ +package corroclient + +import ( + "bytes" + "context" + "encoding/json" + "errors" + "fmt" + "io" + "net/http" +) + +func (c *CorroClient) Exec(ctx context.Context, stmts []Statement) (*ExecResult, error) { + payload, err := json.Marshal(stmts) + if err != nil { + return nil, err + } + + buffer := bytes.NewBuffer(payload) + + request, err := http.NewRequest("POST", c.getURL("/v1/transactions"), buffer) + if err != nil { + return nil, err + } + + resp, err := c.request(request) + if err != nil { + return nil, err + + } + + if resp.StatusCode != http.StatusOK { + bodyErr, _ := io.ReadAll(resp.Body) + return nil, fmt.Errorf("corroclient: invalid status code: %d, body: %s", resp.StatusCode, string(bodyErr)) + } + + var execResult ExecResult + err = json.NewDecoder(resp.Body).Decode(&execResult) + if err != nil { + return nil, err + } + + return &execResult, nil +} + +type ExecResult struct { + Results []Result `json:"results"` +} + +func (e *ExecResult) Errors() []error { + var errs []error + for _, res := range e.Results { + err := res.Err() + errs = append(errs, err) + } + return errs +} + +type Result struct { + Error string `json:"error"` + RowAffected int `json:"rows_affected"` + Time float64 `json:"time"` +} + +func (r *Result) Err() error { + if r.Error != "" { + return errors.New(r.Error) + } + return nil +} diff --git a/go.mod b/go.mod new file mode 100644 index 0000000..93dc939 --- /dev/null +++ b/go.mod @@ -0,0 +1,3 @@ +module github.com/valyentdev/corroclient + +go 1.22.5 diff --git a/queries.go b/queries.go new file mode 100644 index 0000000..e0d8c86 --- /dev/null +++ b/queries.go @@ -0,0 +1,104 @@ +package corroclient + +import ( + "bufio" + "bytes" + "context" + "encoding/json" + "errors" + "fmt" + "io" + "net/http" + "sync" +) + +var ErrNoRows = errors.New("corroclient: no rows") + +func (c *CorroClient) Query(ctx context.Context, stmt Statement) (*Rows, error) { + payload, err := json.Marshal(stmt) + if err != nil { + return nil, err + } + + buffer := bytes.NewBuffer(payload) + + request, err := http.NewRequest("POST", c.getURL("/v1/queries"), buffer) + if err != nil { + return nil, err + } + + resp, err := c.request(request) + if err != nil { + return nil, err + } + defer resp.Body.Close() + + if resp.StatusCode != http.StatusOK { + bodyBytes, _ := io.ReadAll(resp.Body) + return nil, fmt.Errorf("corroclient: invalid status code: %d, body: %s", resp.StatusCode, string(bodyBytes)) + } + + reader := bufio.NewReader(resp.Body) + + var columns Columns + + rows := []*Row{} + + for { + data, err := reader.ReadBytes('\n') + if err != nil { + return nil, err + } + + var e event + if err := json.Unmarshal(data, &e); err != nil { + return nil, err + } + + if e.Columns != nil { + columns = e.Columns + continue + } + + if e.Row != nil { + row, err := readRow(e.Row) + if err != nil { + return nil, err + } + + rows = append(rows, row) + } + + if e.EOQ != nil { + break + } + + } + + if len(rows) == 0 { + return nil, ErrNoRows + } + + return &Rows{ + columns: columns, + rows: rows, + mutex: sync.RWMutex{}, + currentIndex: -1, + }, nil +} + +func (c *CorroClient) QueryRow(ctx context.Context, stmt Statement) (*Row, error) { + rows, err := c.Query(ctx, stmt) + if err != nil { + return nil, err + } + + if !rows.Next() { + return nil, ErrNoRows // should never append but just in case... + } + + row := rows.rows[rows.currentIndex] + row.columns = rows.columns + + return row, nil +} diff --git a/rows.go b/rows.go new file mode 100644 index 0000000..fed6ffe --- /dev/null +++ b/rows.go @@ -0,0 +1,75 @@ +package corroclient + +import ( + "errors" + "fmt" + "sync" +) + +var ErrScan = errors.New("corroclient: scan error") + +// Warning: Scan does not handle time.Time because of the various ways time can be stored +// in SQLite and JSON. You're responsible for converting time.Time yourself from numbers types or +// strings. +func (r *Row) Scan(dest ...any) error { + for i, value := range r.values { + if value == nil { + continue + } + switch v := value.(type) { + case float64: + if err := scanJSONNumber(v, dest[i]); err != nil { + return fmt.Errorf("%w, failed to scan JSON float64 %s", err, value) + } + continue + case string: + if err := scanJSONString(v, dest[i]); err != nil { + return fmt.Errorf("%w, failed to scan JSON string %s", err, value) + } + continue + case bool: + if err := scanJSONBool(v, dest[i]); err != nil { + return fmt.Errorf("%w, failed to scan JSON bool %s", err, value) + } + continue + } + } + return nil +} + +type Rows struct { + columns []string + rows []*Row + currentIndex int + mutex sync.RWMutex +} + +func (r *Rows) Columns() []string { + return r.columns +} + +func (r *Rows) Next() bool { + r.mutex.Lock() + defer r.mutex.Unlock() + if r.currentIndex == len(r.rows)-1 { + return false + } + + r.currentIndex++ + return true +} + +// Warning: Scan does not handle time.Time because of the various ways time can be stored +// in SQLite and JSON. You're responsible for converting time.Time yourself from numbers types or +// strings. +func (r *Rows) Scan(dest ...any) error { + r.mutex.RLock() + defer r.mutex.RUnlock() + if r.currentIndex == -1 { + return fmt.Errorf("you must call Next at least once before calling Scan") + } + + row := r.rows[r.currentIndex] + + return row.Scan(dest...) +} diff --git a/scan.go b/scan.go new file mode 100644 index 0000000..d91c285 --- /dev/null +++ b/scan.go @@ -0,0 +1,74 @@ +package corroclient + +func scanJSONNumber(number float64, d interface{}) error { + switch dest := d.(type) { + case *int: + *dest = int(number) + return nil + case *int8: + *dest = int8(number) + return nil + case *int16: + *dest = int16(number) + return nil + case *int32: + *dest = int32(number) + return nil + case *int64: + *dest = int64(number) + return nil + case *uint: + *dest = uint(number) + return nil + case *uint8: + *dest = uint8(number) + return nil + case *uint16: + *dest = uint16(number) + return nil + case *uint32: + *dest = uint32(number) + return nil + case *uint64: + *dest = uint64(number) + return nil + case *float32: + *dest = float32(number) + return nil + case *float64: + *dest = float64(number) + return nil + case *bool: + *dest = number != 0 + return nil + } + + return ErrScan +} + +func scanJSONString(s string, dest any) error { + if dest == nil { + return ErrScan + } + + switch d := dest.(type) { + case *string: + *d = s + return nil + case *[]byte: + *d = []byte(s) + return nil + } + + return ErrScan +} + +func scanJSONBool(b any, dest interface{}) error { + + switch d := dest.(type) { + case *bool: + *d = b.(bool) + return nil + } + return ErrScan +} diff --git a/subscriptions.go b/subscriptions.go new file mode 100644 index 0000000..67fcd50 --- /dev/null +++ b/subscriptions.go @@ -0,0 +1,191 @@ +package corroclient + +import ( + "bufio" + "bytes" + "context" + "encoding/json" + "errors" + "io" + "net/http" +) + +type event struct { + EOQ *EOQ `json:"eoq"` + Columns Columns `json:"columns"` + Row []any `json:"row"` + Change []any `json:"change"` +} + +func (c *CorroClient) request(req *http.Request) (*http.Response, error) { + if c.bearer != "" { + req.Header.Set("Authorization", c.bearer) + } + req.Header.Set("Content-Type", "application/json") + return c.c.Do(req) +} + +func (c *CorroClient) subscribe(ctx context.Context, body io.ReadCloser) (<-chan Event, error) { + reader := bufio.NewReader(body) + eventChan := make(chan Event) + + go func() { + defer body.Close() + columns := []string{} + for { + eventData, _, err := reader.ReadLine() + if err != nil { + eventChan <- &Error{Message: err.Error()} + break + } + + var e event + + err = json.Unmarshal(eventData, &e) + if err != nil { + eventChan <- &Error{Message: err.Error()} + break + } + + if e.Columns != nil { + columns = e.Columns + continue + } + select { + case <-ctx.Done(): + body.Close() + close(eventChan) + return + default: + eventChan <- readEvent(e, columns) + } + + } + }() + + return eventChan, nil +} + +func (c *CorroClient) PostSubscription(ctx context.Context, statement Statement) (*Subscription, error) { + data, err := json.Marshal(statement) + if err != nil { + return nil, err + } + + req, err := http.NewRequestWithContext(ctx, "POST", c.getURL("/v1/subscriptions"), bytes.NewBuffer(data)) + if err != nil { + return nil, err + } + + resp, err := c.request(req) + if err != nil { + return nil, err + } + + if resp.StatusCode != http.StatusOK { + return nil, errors.New("corrosubs: Invalid status code") + } + + subscriptionId := resp.Header.Get("Corro-Query-Id") + + subCtx, cancel := context.WithCancel(context.Background()) + defer func() { + if err != nil { + cancel() + } + }() + + eventChan, err := c.subscribe(subCtx, resp.Body) + if err != nil { + return nil, err + } + + return &Subscription{ + id: subscriptionId, + ctx: subCtx, + cancel: cancel, + events: eventChan, + }, nil +} + +func (c *CorroClient) GetSubscription(ctx context.Context, subscriptionId string) (*Subscription, error) { + req, err := http.NewRequestWithContext(ctx, "GET", c.getURL("/v1/subscriptions/"+subscriptionId), nil) + if err != nil { + return nil, err + } + + resp, err := c.request(req) + if err != nil { + return nil, err + } + + if resp.StatusCode != http.StatusOK { + return nil, errors.New("corrosubs: Invalid status code") + } + + subCtx, cancel := context.WithCancel(context.Background()) + defer func() { + if err != nil { + cancel() + } + }() + + eventChan, err := c.subscribe(subCtx, resp.Body) + if err != nil { + return nil, err + } + + return &Subscription{ + id: subscriptionId, + ctx: subCtx, + cancel: cancel, + events: eventChan, + }, nil +} + +func readEvent(event event, columns []string) Event { + if event.EOQ != nil { + return event.EOQ + } + if event.Columns != nil { + return event.Columns + } + if event.Row != nil { + row, err := readRow(event.Row) + if err != nil { + return &Error{Message: err.Error()} + } + + row.columns = columns + return row + } + + if event.Change != nil { + change, err := readChange(event.Change) + if err != nil { + return &Error{Message: err.Error()} + } + + return change + } + return &Error{Message: "Unknown event type"} +} + +type Subscription struct { + id string + ctx context.Context + cancel context.CancelFunc + events <-chan Event +} + +func (s *Subscription) Close() { + s.cancel() +} + +func (s *Subscription) Events() <-chan Event { + return s.events +} + +func (s *Subscription) Id() string { + return s.id +} diff --git a/types.go b/types.go new file mode 100644 index 0000000..741e647 --- /dev/null +++ b/types.go @@ -0,0 +1,141 @@ +package corroclient + +import ( + "errors" +) + +type Statement struct { + Query string `json:"query"` + Params []any `json:"params,omitempty"` + NamedParams map[string]any `json:"named_params,omitempty"` +} + +type EventType string + +const ( + EventTypeRow EventType = "row" + EventTypeEOQ EventType = "eoq" + EventTypeChange EventType = "change" + EventTypeColumns EventType = "columns" + EventTypeError EventType = "error" +) + +type Event interface { + Type() EventType +} + +type ChangeType string + +const ( + ChangeTypeInsert ChangeType = "insert" + ChangeTypeUpdate ChangeType = "update" + ChangeTypeDelete ChangeType = "delete" +) + +type Change struct { + ChangeId int64 `json:"change_id"` + ChangeType ChangeType `json:"change_type"` + Row *Row `json:"row"` +} + +func (c *Change) Type() EventType { + return EventTypeChange +} + +type Row struct { + rowId int64 + values []any + columns []string +} + +func (r *Row) RowId() int64 { + return r.rowId +} + +func (r *Row) Type() EventType { + return EventTypeRow +} + +type EOQ struct { + ChangeId int64 `json:"change_id"` + Time float64 `json:"time"` +} + +func (e *EOQ) Type() EventType { + return EventTypeEOQ +} + +type Columns []string + +func (c Columns) Type() EventType { + return EventTypeColumns +} + +type Error struct { + Message string `json:"message"` +} + +func (e *Error) Type() EventType { + return EventTypeError +} + +var ErrInvalidRow = errors.New("corrosubs: Invalid row") + +func readRow(data []any) (*Row, error) { + if len(data) != 2 { + return nil, ErrInvalidRow + } + + rowIdFloat, ok := data[0].(float64) + if !ok { + return nil, ErrInvalidRow + } + + rowId := int64(rowIdFloat) + values, ok := data[1].([]any) + if !ok { + return nil, ErrInvalidRow + } + + return &Row{ + rowId: rowId, + values: values, + }, nil +} + +var ErrInvalidChange = errors.New("corrosubs: Invalid change") + +func readChange(data []any) (*Change, error) { + if len(data) != 4 { + return nil, ErrInvalidRow + } + + changeType, ok := data[0].(string) + if !ok { + return nil, ErrInvalidChange + } + + rowId, ok := data[1].(float64) + if !ok { + return nil, ErrInvalidChange + } + + values, ok := data[2].([]any) + if !ok { + return nil, ErrInvalidChange + } + + changeId, ok := data[3].(float64) + if !ok { + return nil, ErrInvalidChange + } + + return &Change{ + ChangeId: int64(changeId), + ChangeType: ChangeType(changeType), + Row: &Row{ + rowId: int64(rowId), + values: values, + }, + }, nil +}