add shreder sell decode

pkg/shreder/client.go

@@ -2,37 +2,34 @@ package shreder
 
 import (
 	"context"
+	"log/slog"
 
-	"github.com/samlior/libsam/pkg/logger"
-	"github.com/samlior/libsam/pkg/txparser"
-	"github.com/samlior/libsam/pkg/types"
-	"github.com/samlior/libsam/third_party/shreder_protos"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/credentials/insecure"
 )
 
-type ShrederClient struct {
-	log logger.Logger
+type Client struct {
+	log *slog.Logger
 
 	conn         *grpc.ClientConn
-	client       shreder_protos.ShrederServiceClient
-	subscription map[string]*shreder_protos.SubscribeRequestFilterTransactions
+	client       ShrederServiceClient
+	subscription map[string]*SubscribeRequestFilterTransactions
 }
 
 func NewShrederClient(
-	logger logger.Logger,
 	url string,
-	subscription map[string]*shreder_protos.SubscribeRequestFilterTransactions,
-) (*ShrederClient, func(), error) {
+	subscription map[string]*SubscribeRequestFilterTransactions,
+) (*Client, func(), error) {
 	conn, err := grpc.NewClient(url, grpc.WithTransportCredentials(insecure.NewCredentials()))
 	if err != nil {
 		return nil, func() {}, err
 	}
 
-	s := &ShrederClient{
+	logger := slog.Default()
+	s := &Client{
 		log:          logger,
 		conn:         conn,
-		client:       shreder_protos.NewShrederServiceClient(conn),
+		client:       NewShrederServiceClient(conn),
 		subscription: subscription,
 	}
 
@@ -41,24 +38,24 @@ func NewShrederClient(
 	}, nil
 }
 
-func (c *ShrederClient) Wait() {
+func (c *Client) Wait() {
 	c.log.Debug("waiting for shreder client to stop")
 
 	err := c.conn.Close()
 	if err != nil {
-		c.log.Errorf("failed to close connection: %v", err)
+		c.log.Error("failed to close connection: ", "err", err)
 	}
 
 	c.log.Debug("shreder client stopped")
 }
 
-func (c *ShrederClient) ReadSync(ctx context.Context, txCh chan<- types.TxSignalBatch) error {
+func (c *Client) ReadSync(ctx context.Context, txCh chan<- TxSignalBatch) error {
 	stream, err := c.client.SubscribeTransactions(ctx)
 	if err != nil {
 		return err
 	}
 
-	err = stream.Send(&shreder_protos.SubscribeTransactionsRequest{
+	err = stream.Send(&SubscribeTransactionsRequest{
 		Transactions: c.subscription,
 	})
 	if err != nil {
@@ -71,7 +68,7 @@ func (c *ShrederClient) ReadSync(ctx context.Context, txCh chan<- types.TxSignal
 			return err
 		}
 
-		txBatch := txparser.ParseTransaction(response.Transaction)
+		txBatch := ParseTransaction(response.Transaction)
 		if len(txBatch) == 0 {
 			continue
 		}

pkg/shreder/shreder.proto

@@ -0,0 +1,76 @@
+syntax = "proto3";
+
+package shredstream;
+import "google/protobuf/timestamp.proto";
+
+option go_package = "github.com/samlior/libsam/pkg//shreder";
+
+service ShrederService {
+  rpc SubscribeEntries(SubscribeEntriesRequest) returns (stream Entry);
+  rpc SubscribeTransactions(stream SubscribeTransactionsRequest) returns (stream SubscribeTransactionsResponse);
+}
+
+message SubscribeEntriesRequest {
+  // tbd: we may want to add filters here
+}
+
+message SubscribeTransactionsRequest {
+  map<string, SubscribeRequestFilterTransactions> transactions = 3;
+}
+
+message SubscribeTransactionsResponse {
+  repeated string filters = 1;
+  SubscribeUpdateTransaction transaction = 4;
+  google.protobuf.Timestamp created_at = 11;
+}
+
+message SubscribeUpdateTransaction {
+  Transaction transaction = 1;
+  uint64 slot = 2;
+}
+
+message SubscribeRequestFilterTransactions {
+  repeated string account_include = 3;
+  repeated string account_exclude = 4;
+  repeated string account_required = 6;
+}
+
+message Entry {
+  // the slot that the entry is from
+  uint64 slot = 1;
+
+  // Serialized bytes of Vec<Entry>: https://docs.rs/solana-entry/latest/solana_entry/entry/struct.Entry.html
+  bytes entries = 2;
+}
+
+message MessageHeader {
+  uint32 num_required_signatures = 1;
+  uint32 num_readonly_signed_accounts = 2;
+  uint32 num_readonly_unsigned_accounts = 3;
+}
+
+message CompiledInstruction {
+  uint32 program_id_index = 1;
+  bytes accounts = 2;
+  bytes data = 3;
+}
+
+message MessageAddressTableLookup {
+  bytes account_key = 1;
+  bytes writable_indexes = 2;
+  bytes readonly_indexes = 3;
+}
+
+message Message {
+  MessageHeader header = 1;
+  repeated bytes account_keys = 2;
+  bytes recent_blockhash = 3;
+  repeated CompiledInstruction instructions = 4;
+  bool versioned = 5;
+  repeated MessageAddressTableLookup address_table_lookups = 6;
+}
+
+message Transaction {
+  repeated bytes signatures = 1;
+  Message message = 2;
+}

pkg/shreder/shreder_grpc.pb.go

@@ -0,0 +1,205 @@
+// Code generated by protoc-gen-go-grpc. DO NOT EDIT.
+// versions:
+// - protoc-gen-go-grpc v1.3.0
+// - protoc             v6.33.1
+// source: shreder.proto
+
+package shreder
+
+import (
+	context "context"
+	grpc "google.golang.org/grpc"
+	codes "google.golang.org/grpc/codes"
+	status "google.golang.org/grpc/status"
+)
+
+// This is a compile-time assertion to ensure that this generated file
+// is compatible with the grpc package it is being compiled against.
+// Requires gRPC-Go v1.32.0 or later.
+const _ = grpc.SupportPackageIsVersion7
+
+const (
+	ShrederService_SubscribeEntries_FullMethodName      = "/shredstream.ShrederService/SubscribeEntries"
+	ShrederService_SubscribeTransactions_FullMethodName = "/shredstream.ShrederService/SubscribeTransactions"
+)
+
+// ShrederServiceClient is the client API for ShrederService service.
+//
+// For semantics around ctx use and closing/ending streaming RPCs, please refer to https://pkg.go.dev/google.golang.org/grpc/?tab=doc#ClientConn.NewStream.
+type ShrederServiceClient interface {
+	SubscribeEntries(ctx context.Context, in *SubscribeEntriesRequest, opts ...grpc.CallOption) (ShrederService_SubscribeEntriesClient, error)
+	SubscribeTransactions(ctx context.Context, opts ...grpc.CallOption) (ShrederService_SubscribeTransactionsClient, error)
+}
+
+type shrederServiceClient struct {
+	cc grpc.ClientConnInterface
+}
+
+func NewShrederServiceClient(cc grpc.ClientConnInterface) ShrederServiceClient {
+	return &shrederServiceClient{cc}
+}
+
+func (c *shrederServiceClient) SubscribeEntries(ctx context.Context, in *SubscribeEntriesRequest, opts ...grpc.CallOption) (ShrederService_SubscribeEntriesClient, error) {
+	stream, err := c.cc.NewStream(ctx, &ShrederService_ServiceDesc.Streams[0], ShrederService_SubscribeEntries_FullMethodName, opts...)
+	if err != nil {
+		return nil, err
+	}
+	x := &shrederServiceSubscribeEntriesClient{stream}
+	if err := x.ClientStream.SendMsg(in); err != nil {
+		return nil, err
+	}
+	if err := x.ClientStream.CloseSend(); err != nil {
+		return nil, err
+	}
+	return x, nil
+}
+
+type ShrederService_SubscribeEntriesClient interface {
+	Recv() (*Entry, error)
+	grpc.ClientStream
+}
+
+type shrederServiceSubscribeEntriesClient struct {
+	grpc.ClientStream
+}
+
+func (x *shrederServiceSubscribeEntriesClient) Recv() (*Entry, error) {
+	m := new(Entry)
+	if err := x.ClientStream.RecvMsg(m); err != nil {
+		return nil, err
+	}
+	return m, nil
+}
+
+func (c *shrederServiceClient) SubscribeTransactions(ctx context.Context, opts ...grpc.CallOption) (ShrederService_SubscribeTransactionsClient, error) {
+	stream, err := c.cc.NewStream(ctx, &ShrederService_ServiceDesc.Streams[1], ShrederService_SubscribeTransactions_FullMethodName, opts...)
+	if err != nil {
+		return nil, err
+	}
+	x := &shrederServiceSubscribeTransactionsClient{stream}
+	return x, nil
+}
+
+type ShrederService_SubscribeTransactionsClient interface {
+	Send(*SubscribeTransactionsRequest) error
+	Recv() (*SubscribeTransactionsResponse, error)
+	grpc.ClientStream
+}
+
+type shrederServiceSubscribeTransactionsClient struct {
+	grpc.ClientStream
+}
+
+func (x *shrederServiceSubscribeTransactionsClient) Send(m *SubscribeTransactionsRequest) error {
+	return x.ClientStream.SendMsg(m)
+}
+
+func (x *shrederServiceSubscribeTransactionsClient) Recv() (*SubscribeTransactionsResponse, error) {
+	m := new(SubscribeTransactionsResponse)
+	if err := x.ClientStream.RecvMsg(m); err != nil {
+		return nil, err
+	}
+	return m, nil
+}
+
+// ShrederServiceServer is the server API for ShrederService service.
+// All implementations must embed UnimplementedShrederServiceServer
+// for forward compatibility
+type ShrederServiceServer interface {
+	SubscribeEntries(*SubscribeEntriesRequest, ShrederService_SubscribeEntriesServer) error
+	SubscribeTransactions(ShrederService_SubscribeTransactionsServer) error
+	mustEmbedUnimplementedShrederServiceServer()
+}
+
+// UnimplementedShrederServiceServer must be embedded to have forward compatible implementations.
+type UnimplementedShrederServiceServer struct {
+}
+
+func (UnimplementedShrederServiceServer) SubscribeEntries(*SubscribeEntriesRequest, ShrederService_SubscribeEntriesServer) error {
+	return status.Errorf(codes.Unimplemented, "method SubscribeEntries not implemented")
+}
+func (UnimplementedShrederServiceServer) SubscribeTransactions(ShrederService_SubscribeTransactionsServer) error {
+	return status.Errorf(codes.Unimplemented, "method SubscribeTransactions not implemented")
+}
+func (UnimplementedShrederServiceServer) mustEmbedUnimplementedShrederServiceServer() {}
+
+// UnsafeShrederServiceServer may be embedded to opt out of forward compatibility for this service.
+// Use of this interface is not recommended, as added methods to ShrederServiceServer will
+// result in compilation errors.
+type UnsafeShrederServiceServer interface {
+	mustEmbedUnimplementedShrederServiceServer()
+}
+
+func RegisterShrederServiceServer(s grpc.ServiceRegistrar, srv ShrederServiceServer) {
+	s.RegisterService(&ShrederService_ServiceDesc, srv)
+}
+
+func _ShrederService_SubscribeEntries_Handler(srv interface{}, stream grpc.ServerStream) error {
+	m := new(SubscribeEntriesRequest)
+	if err := stream.RecvMsg(m); err != nil {
+		return err
+	}
+	return srv.(ShrederServiceServer).SubscribeEntries(m, &shrederServiceSubscribeEntriesServer{stream})
+}
+
+type ShrederService_SubscribeEntriesServer interface {
+	Send(*Entry) error
+	grpc.ServerStream
+}
+
+type shrederServiceSubscribeEntriesServer struct {
+	grpc.ServerStream
+}
+
+func (x *shrederServiceSubscribeEntriesServer) Send(m *Entry) error {
+	return x.ServerStream.SendMsg(m)
+}
+
+func _ShrederService_SubscribeTransactions_Handler(srv interface{}, stream grpc.ServerStream) error {
+	return srv.(ShrederServiceServer).SubscribeTransactions(&shrederServiceSubscribeTransactionsServer{stream})
+}
+
+type ShrederService_SubscribeTransactionsServer interface {
+	Send(*SubscribeTransactionsResponse) error
+	Recv() (*SubscribeTransactionsRequest, error)
+	grpc.ServerStream
+}
+
+type shrederServiceSubscribeTransactionsServer struct {
+	grpc.ServerStream
+}
+
+func (x *shrederServiceSubscribeTransactionsServer) Send(m *SubscribeTransactionsResponse) error {
+	return x.ServerStream.SendMsg(m)
+}
+
+func (x *shrederServiceSubscribeTransactionsServer) Recv() (*SubscribeTransactionsRequest, error) {
+	m := new(SubscribeTransactionsRequest)
+	if err := x.ServerStream.RecvMsg(m); err != nil {
+		return nil, err
+	}
+	return m, nil
+}
+
+// ShrederService_ServiceDesc is the grpc.ServiceDesc for ShrederService service.
+// It's only intended for direct use with grpc.RegisterService,
+// and not to be introspected or modified (even as a copy)
+var ShrederService_ServiceDesc = grpc.ServiceDesc{
+	ServiceName: "shredstream.ShrederService",
+	HandlerType: (*ShrederServiceServer)(nil),
+	Methods:     []grpc.MethodDesc{},
+	Streams: []grpc.StreamDesc{
+		{
+			StreamName:    "SubscribeEntries",
+			Handler:       _ShrederService_SubscribeEntries_Handler,
+			ServerStreams: true,
+		},
+		{
+			StreamName:    "SubscribeTransactions",
+			Handler:       _ShrederService_SubscribeTransactions_Handler,
+			ServerStreams: true,
+			ClientStreams: true,
+		},
+	},
+	Metadata: "shreder.proto",
+}

pkg/shreder/tx.go

@@ -0,0 +1,45 @@
+package shreder
+
+import (
+	"time"
+
+	"github.com/shopspring/decimal"
+)
+
+const (
+	TokenDecimals = 6
+	SolDecimals   = 9
+)
+
+type TxSignal struct {
+	Source        string          `json:"source"`
+	TxHash        string          `json:"tx_hash"`
+	Maker         string          `json:"maker"`
+	Token0Address string          `json:"token0_address"`
+	Token1Address string          `json:"token1_address"`
+	Token0Amount  decimal.Decimal `json:"token0_amount"`
+	Token1Amount  decimal.Decimal `json:"token1_amount"`
+	Block         uint64          `json:"block"`
+	BlockAt       time.Time       `json:"block_at"`
+	BlockIndex    uint64          `json:"block_index"`
+	Event         string          `json:"event"`
+	Program       string          `json:"program"`
+	IsProcessed   bool            `json:"is_processed"`
+	IsToken2022   bool            `json:"is_token2022"`
+	IsMayhemMode  bool            `json:"is_mayhem_mode"`
+	TxFee         decimal.Decimal `json:"tx_fee"`
+
+	ExactSOL bool `json:"exact_in"`
+
+	// parsed values
+	Token0AmountUint64 uint64 `json:"-"`
+	Token1AmountUint64 uint64 `json:"-"`
+}
+
+func (t *TxSignal) Parse() *TxSignal {
+	t.Token0AmountUint64 = t.Token0Amount.Mul(decimal.New(1, TokenDecimals)).BigInt().Uint64()
+	t.Token1AmountUint64 = t.Token1Amount.Mul(decimal.New(1, SolDecimals)).BigInt().Uint64()
+	return t
+}
+
+type TxSignalBatch = []*TxSignal