chrissewell/hetty
1
0
Fork 0
mirror of https://github.com/dstotijn/hetty.git synced 2025-07-01 18:47:29 -04:00
Code Issues Packages Projects Releases Wiki Activity

Replace GraphQL server with Connect RPC

Browse Source
This commit is contained in:
David Stotijn
2025-02-05 21:54:59 +01:00
parent 52c83a1989
commit 6889c9c183
53 changed files with 5875 additions and 11685 deletions
Download Patch File Download Diff File Expand all files Collapse all files

69
pkg/db/bolt/sender.go
View File

@ -1,16 +1,13 @@
package bolt
import (
"bytes"
"context"
"encoding/gob"
"errors"
"fmt"
"github.com/oklog/ulid"
bolt "go.etcd.io/bbolt"
"google.golang.org/protobuf/proto"
"github.com/dstotijn/hetty/pkg/scope"
"github.com/dstotijn/hetty/pkg/sender"
)
@ -18,8 +15,8 @@ var ErrSenderRequestsBucketNotFound = errors.New("bolt: sender requests bucket n
var senderReqsBucketName = []byte("sender_requests")
func senderReqsBucket(tx *bolt.Tx, projectID ulid.ULID) (*bolt.Bucket, error) {
pb, err := projectBucket(tx, projectID[:])
func senderReqsBucket(tx *bolt.Tx, projectID string) (*bolt.Bucket, error) {
pb, err := projectBucket(tx, projectID)
if err != nil {
return nil, err
}
@ -32,21 +29,19 @@ func senderReqsBucket(tx *bolt.Tx, projectID ulid.ULID) (*bolt.Bucket, error) {
return b, nil
}
func (db *Database) StoreSenderRequest(ctx context.Context, req sender.Request) error {
buf := bytes.Buffer{}
err := gob.NewEncoder(&buf).Encode(req)
func (db *Database) StoreSenderRequest(ctx context.Context, req *sender.Request) error {
rawReq, err := proto.Marshal(req)
if err != nil {
return fmt.Errorf("bolt: failed to encode sender request: %w", err)
return fmt.Errorf("bolt: failed to marshal sender request: %w", err)
}
err = db.bolt.Update(func(tx *bolt.Tx) error {
senderReqsBucket, err := senderReqsBucket(tx, req.ProjectID)
senderReqsBucket, err := senderReqsBucket(tx, req.ProjectId)
if err != nil {
return fmt.Errorf("failed to get sender requests bucket: %w", err)
}
err = senderReqsBucket.Put(req.ID[:], buf.Bytes())
err = senderReqsBucket.Put([]byte(req.Id), rawReq)
if err != nil {
return fmt.Errorf("failed to put sender request: %w", err)
}
@ -60,9 +55,9 @@ func (db *Database) StoreSenderRequest(ctx context.Context, req sender.Request)
return nil
}
func (db *Database) FindSenderRequestByID(ctx context.Context, projectID, senderReqID ulid.ULID) (req sender.Request, err error) {
if projectID.Compare(ulid.ULID{}) == 0 {
return sender.Request{}, sender.ErrProjectIDMustBeSet
func (db *Database) FindSenderRequestByID(ctx context.Context, projectID, senderReqID string) (req *sender.Request, err error) {
if projectID == "" {
return nil, sender.ErrProjectIDMustBeSet
}
err = db.bolt.View(func(tx *bolt.Tx) error {
@ -71,63 +66,49 @@ func (db *Database) FindSenderRequestByID(ctx context.Context, projectID, sender
return fmt.Errorf("failed to get sender requests bucket: %w", err)
}
rawSenderReq := senderReqsBucket.Get(senderReqID[:])
rawSenderReq := senderReqsBucket.Get([]byte(senderReqID))
if rawSenderReq == nil {
return sender.ErrRequestNotFound
}
err = gob.NewDecoder(bytes.NewReader(rawSenderReq)).Decode(&req)
req = &sender.Request{}
err = proto.Unmarshal(rawSenderReq, req)
if err != nil {
return fmt.Errorf("failed to decode sender request: %w", err)
return fmt.Errorf("failed to unmarshal sender request: %w", err)
}
return nil
})
if err != nil {
return sender.Request{}, fmt.Errorf("bolt: failed to commit transaction: %w", err)
return nil, fmt.Errorf("bolt: failed to commit transaction: %w", err)
}
return req, nil
}
func (db *Database) FindSenderRequests(ctx context.Context, filter sender.FindRequestsFilter, scope *scope.Scope) (reqs []sender.Request, err error) {
if filter.ProjectID.Compare(ulid.ULID{}) == 0 {
return nil, sender.ErrProjectIDMustBeSet
}
func (db *Database) FindSenderRequests(ctx context.Context, projectID string, filterFn func(req *sender.Request) (bool, error)) (reqs []*sender.Request, err error) {
tx, err := db.bolt.Begin(false)
if err != nil {
return nil, fmt.Errorf("bolt: failed to begin transaction: %w", err)
}
defer tx.Rollback()
b, err := senderReqsBucket(tx, filter.ProjectID)
b, err := senderReqsBucket(tx, projectID)
if err != nil {
return nil, fmt.Errorf("failed to get sender requests bucket: %w", err)
}
err = b.ForEach(func(senderReqID, rawSenderReq []byte) error {
var req sender.Request
err = gob.NewDecoder(bytes.NewReader(rawSenderReq)).Decode(&req)
req := &sender.Request{}
err = proto.Unmarshal(rawSenderReq, req)
if err != nil {
return fmt.Errorf("failed to decode sender request: %w", err)
return fmt.Errorf("failed to unmarshal sender request: %w", err)
}
if filter.OnlyInScope {
if !req.MatchScope(scope) {
return nil
}
}
// Filter by search expression. TODO: Once pagination is introduced,
// this filter logic should be done as items are retrieved.
if filter.SearchExpr != nil {
match, err := req.Matches(filter.SearchExpr)
if filterFn != nil {
match, err := filterFn(req)
if err != nil {
return fmt.Errorf(
"bolt: failed to match search expression for sender request (id: %v): %w",
senderReqID, err,
)
return fmt.Errorf("failed to filter sender request: %w", err)
}
if !match {
@ -150,7 +131,7 @@ func (db *Database) FindSenderRequests(ctx context.Context, filter sender.FindRe
return reqs, nil
}
func (db *Database) DeleteSenderRequests(ctx context.Context, projectID ulid.ULID) error {
func (db *Database) DeleteSenderRequests(ctx context.Context, projectID string) error {
err := db.bolt.Update(func(tx *bolt.Tx) error {
senderReqsBucket, err := senderReqsBucket(tx, projectID)
if err != nil {