jacob-mathison/query-orchestration
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Activity

Merged in feature/baseTemplate (pull request #1)

Browse Source 
feat!: base template

* first bit of templating

* codeowners

* linuxbased

* restart

* baseline project

* add grpc api and basic integration test

* startqueue

* queueMsg

* splitscripts

* migrations

* queueintegrationtest

* gateway
This commit is contained in:
Michael McGuinness
2024-12-06 14:38:42 +00:00
parent cccbdaad4f
commit b1f8ac453b
2845 changed files with 832056 additions and 50 deletions
Download Patch File Download Diff File Expand all files Collapse all files
api/grpc/controllers/name.go
+40
View File
@@ -0,0 +1,40 @@
package controllers
import (
"context"
"gotemplate/api/grpc/spec"
"gotemplate/internal/name"
)
type NameController struct {
spec.UnimplementedNameServer
name name.Service
}
func NewNameController(nameSvc name.Service) *NameController {
return &NameController{
name: nameSvc,
}
}
func (s *NameController) Add(ctx context.Context, req *spec.NameMessage) (*spec.NameMessage, error) {
err := s.name.Add(ctx, req.GetName())
if err != nil {
return &spec.NameMessage{}, err
}
return &spec.NameMessage{
Name: req.GetName(),
}, nil
}
func (s *NameController) Get(ctx context.Context, req *spec.NameMessage) (*spec.NameMessage, error) {
name, err := s.name.Get(ctx, req.GetName())
if err != nil {
return &spec.NameMessage{}, err
}
return &spec.NameMessage{
Name: name,
}, nil
}
api/grpc/protobuf/main.proto
+14
View File
@@ -0,0 +1,14 @@
syntax = "proto3";
package gotemplate;
option go_package = "./spec";
service Name {
rpc Add(NameMessage) returns (NameMessage);
rpc Get(NameMessage) returns (NameMessage);
}
message NameMessage {
string name = 1;
}
api/grpc/spec/main.pb.go
+137
View File
@@ -0,0 +1,137 @@
// Code generated by protoc-gen-go. DO NOT EDIT.
// versions:
// protoc-gen-go v1.35.2
// protoc v3.12.4
// source: main.proto
package spec
import (
protoreflect "google.golang.org/protobuf/reflect/protoreflect"
protoimpl "google.golang.org/protobuf/runtime/protoimpl"
reflect "reflect"
sync "sync"
)
const (
// Verify that this generated code is sufficiently up-to-date.
_ = protoimpl.EnforceVersion(20 - protoimpl.MinVersion)
// Verify that runtime/protoimpl is sufficiently up-to-date.
_ = protoimpl.EnforceVersion(protoimpl.MaxVersion - 20)
)
type NameMessage struct {
state protoimpl.MessageState
sizeCache protoimpl.SizeCache
unknownFields protoimpl.UnknownFields
Name string `protobuf:"bytes,1,opt,name=name,proto3" json:"name,omitempty"`
}
func (x *NameMessage) Reset() {
*x = NameMessage{}
mi := &file_main_proto_msgTypes[0]
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
ms.StoreMessageInfo(mi)
}
func (x *NameMessage) String() string {
return protoimpl.X.MessageStringOf(x)
}
func (*NameMessage) ProtoMessage() {}
func (x *NameMessage) ProtoReflect() protoreflect.Message {
mi := &file_main_proto_msgTypes[0]
if x != nil {
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
if ms.LoadMessageInfo() == nil {
ms.StoreMessageInfo(mi)
}
return ms
}
return mi.MessageOf(x)
}
// Deprecated: Use NameMessage.ProtoReflect.Descriptor instead.
func (*NameMessage) Descriptor() ([]byte, []int) {
return file_main_proto_rawDescGZIP(), []int{0}
}
func (x *NameMessage) GetName() string {
if x != nil {
return x.Name
}
return ""
}
var File_main_proto protoreflect.FileDescriptor
var file_main_proto_rawDesc = []byte{
0x0a, 0x0a, 0x6d, 0x61, 0x69, 0x6e, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x12, 0x0a, 0x67, 0x6f,
0x74, 0x65, 0x6d, 0x70, 0x6c, 0x61, 0x74, 0x65, 0x22, 0x21, 0x0a, 0x0b, 0x4e, 0x61, 0x6d, 0x65,
0x4d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x12, 0x12, 0x0a, 0x04, 0x6e, 0x61, 0x6d, 0x65, 0x18,
0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x04, 0x6e, 0x61, 0x6d, 0x65, 0x32, 0x78, 0x0a, 0x04, 0x4e,
0x61, 0x6d, 0x65, 0x12, 0x37, 0x0a, 0x03, 0x41, 0x64, 0x64, 0x12, 0x17, 0x2e, 0x67, 0x6f, 0x74,
0x65, 0x6d, 0x70, 0x6c, 0x61, 0x74, 0x65, 0x2e, 0x4e, 0x61, 0x6d, 0x65, 0x4d, 0x65, 0x73, 0x73,
0x61, 0x67, 0x65, 0x1a, 0x17, 0x2e, 0x67, 0x6f, 0x74, 0x65, 0x6d, 0x70, 0x6c, 0x61, 0x74, 0x65,
0x2e, 0x4e, 0x61, 0x6d, 0x65, 0x4d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x12, 0x37, 0x0a, 0x03,
0x47, 0x65, 0x74, 0x12, 0x17, 0x2e, 0x67, 0x6f, 0x74, 0x65, 0x6d, 0x70, 0x6c, 0x61, 0x74, 0x65,
0x2e, 0x4e, 0x61, 0x6d, 0x65, 0x4d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x1a, 0x17, 0x2e, 0x67,
0x6f, 0x74, 0x65, 0x6d, 0x70, 0x6c, 0x61, 0x74, 0x65, 0x2e, 0x4e, 0x61, 0x6d, 0x65, 0x4d, 0x65,
0x73, 0x73, 0x61, 0x67, 0x65, 0x42, 0x08, 0x5a, 0x06, 0x2e, 0x2f, 0x73, 0x70, 0x65, 0x63, 0x62,
0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33,
}
var (
file_main_proto_rawDescOnce sync.Once
file_main_proto_rawDescData = file_main_proto_rawDesc
)
func file_main_proto_rawDescGZIP() []byte {
file_main_proto_rawDescOnce.Do(func() {
file_main_proto_rawDescData = protoimpl.X.CompressGZIP(file_main_proto_rawDescData)
})
return file_main_proto_rawDescData
}
var file_main_proto_msgTypes = make([]protoimpl.MessageInfo, 1)
var file_main_proto_goTypes = []any{
(*NameMessage)(nil), // 0: gotemplate.NameMessage
}
var file_main_proto_depIdxs = []int32{
0, // 0: gotemplate.Name.Add:input_type -> gotemplate.NameMessage
0, // 1: gotemplate.Name.Get:input_type -> gotemplate.NameMessage
0, // 2: gotemplate.Name.Add:output_type -> gotemplate.NameMessage
0, // 3: gotemplate.Name.Get:output_type -> gotemplate.NameMessage
2, // [2:4] is the sub-list for method output_type
0, // [0:2] is the sub-list for method input_type
0, // [0:0] is the sub-list for extension type_name
0, // [0:0] is the sub-list for extension extendee
0, // [0:0] is the sub-list for field type_name
}
func init() { file_main_proto_init() }
func file_main_proto_init() {
if File_main_proto != nil {
return
}
type x struct{}
out := protoimpl.TypeBuilder{
File: protoimpl.DescBuilder{
GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
RawDescriptor: file_main_proto_rawDesc,
NumEnums: 0,
NumMessages: 1,
NumExtensions: 0,
NumServices: 1,
},
GoTypes: file_main_proto_goTypes,
DependencyIndexes: file_main_proto_depIdxs,
MessageInfos: file_main_proto_msgTypes,
}.Build()
File_main_proto = out.File
file_main_proto_rawDesc = nil
file_main_proto_goTypes = nil
file_main_proto_depIdxs = nil
}
api/grpc/spec/main_grpc.pb.go
+159
View File
@@ -0,0 +1,159 @@
// Code generated by protoc-gen-go-grpc. DO NOT EDIT.
// versions:
// - protoc-gen-go-grpc v1.5.1
// - protoc v3.12.4
// source: main.proto
package spec
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.64.0 or later.
const _ = grpc.SupportPackageIsVersion9
const (
Name_Add_FullMethodName = "/gotemplate.Name/Add"
Name_Get_FullMethodName = "/gotemplate.Name/Get"
)
// NameClient is the client API for Name 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 NameClient interface {
Add(ctx context.Context, in *NameMessage, opts ...grpc.CallOption) (*NameMessage, error)
Get(ctx context.Context, in *NameMessage, opts ...grpc.CallOption) (*NameMessage, error)
}
type nameClient struct {
cc grpc.ClientConnInterface
}
func NewNameClient(cc grpc.ClientConnInterface) NameClient {
return &nameClient{cc}
}
func (c *nameClient) Add(ctx context.Context, in *NameMessage, opts ...grpc.CallOption) (*NameMessage, error) {
cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...)
out := new(NameMessage)
err := c.cc.Invoke(ctx, Name_Add_FullMethodName, in, out, cOpts...)
if err != nil {
return nil, err
}
return out, nil
}
func (c *nameClient) Get(ctx context.Context, in *NameMessage, opts ...grpc.CallOption) (*NameMessage, error) {
cOpts := append([]grpc.CallOption{grpc.StaticMethod()}, opts...)
out := new(NameMessage)
err := c.cc.Invoke(ctx, Name_Get_FullMethodName, in, out, cOpts...)
if err != nil {
return nil, err
}
return out, nil
}
// NameServer is the server API for Name service.
// All implementations must embed UnimplementedNameServer
// for forward compatibility.
type NameServer interface {
Add(context.Context, *NameMessage) (*NameMessage, error)
Get(context.Context, *NameMessage) (*NameMessage, error)
mustEmbedUnimplementedNameServer()
}
// UnimplementedNameServer must be embedded to have
// forward compatible implementations.
//
// NOTE: this should be embedded by value instead of pointer to avoid a nil
// pointer dereference when methods are called.
type UnimplementedNameServer struct{}
func (UnimplementedNameServer) Add(context.Context, *NameMessage) (*NameMessage, error) {
return nil, status.Errorf(codes.Unimplemented, "method Add not implemented")
}
func (UnimplementedNameServer) Get(context.Context, *NameMessage) (*NameMessage, error) {
return nil, status.Errorf(codes.Unimplemented, "method Get not implemented")
}
func (UnimplementedNameServer) mustEmbedUnimplementedNameServer() {}
func (UnimplementedNameServer) testEmbeddedByValue() {}
// UnsafeNameServer may be embedded to opt out of forward compatibility for this service.
// Use of this interface is not recommended, as added methods to NameServer will
// result in compilation errors.
type UnsafeNameServer interface {
mustEmbedUnimplementedNameServer()
}
func RegisterNameServer(s grpc.ServiceRegistrar, srv NameServer) {
// If the following call pancis, it indicates UnimplementedNameServer was
// embedded by pointer and is nil. This will cause panics if an
// unimplemented method is ever invoked, so we test this at initialization
// time to prevent it from happening at runtime later due to I/O.
if t, ok := srv.(interface{ testEmbeddedByValue() }); ok {
t.testEmbeddedByValue()
}
s.RegisterService(&Name_ServiceDesc, srv)
}
func _Name_Add_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
in := new(NameMessage)
if err := dec(in); err != nil {
return nil, err
}
if interceptor == nil {
return srv.(NameServer).Add(ctx, in)
}
info := &grpc.UnaryServerInfo{
Server: srv,
FullMethod: Name_Add_FullMethodName,
}
handler := func(ctx context.Context, req interface{}) (interface{}, error) {
return srv.(NameServer).Add(ctx, req.(*NameMessage))
}
return interceptor(ctx, in, info, handler)
}
func _Name_Get_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
in := new(NameMessage)
if err := dec(in); err != nil {
return nil, err
}
if interceptor == nil {
return srv.(NameServer).Get(ctx, in)
}
info := &grpc.UnaryServerInfo{
Server: srv,
FullMethod: Name_Get_FullMethodName,
}
handler := func(ctx context.Context, req interface{}) (interface{}, error) {
return srv.(NameServer).Get(ctx, req.(*NameMessage))
}
return interceptor(ctx, in, info, handler)
}
// Name_ServiceDesc is the grpc.ServiceDesc for Name service.
// It's only intended for direct use with grpc.RegisterService,
// and not to be introspected or modified (even as a copy)
var Name_ServiceDesc = grpc.ServiceDesc{
ServiceName: "gotemplate.Name",
HandlerType: (*NameServer)(nil),
Methods: []grpc.MethodDesc{
{
MethodName: "Add",
Handler: _Name_Add_Handler,
},
{
MethodName: "Get",
Handler: _Name_Get_Handler,
},
},
Streams: []grpc.StreamDesc{},
Metadata: "main.proto",
}
api/queue/name.go
+65
View File
@@ -0,0 +1,65 @@
package queue
import (
"context"
"encoding/json"
"gotemplate/internal/name"
"log"
"github.com/aws/aws-sdk-go-v2/aws"
"github.com/aws/aws-sdk-go-v2/service/sqs"
"github.com/aws/aws-sdk-go-v2/service/sqs/types"
)
type NameController struct {
name name.Service
}
func NewNameController(nameSvc name.Service) *NameController {
return &NameController{
name: nameSvc,
}
}
type CreateNameBody struct {
Name string `json:"name"`
}
func (s *NameController) Create(ctx context.Context, config *QueueConfig, msg *types.Message) error {
var body CreateNameBody
err := json.Unmarshal([]byte(*msg.Body), &body)
if err != nil {
return err
}
err = s.name.Add(ctx, body.Name)
if err != nil {
return err
}
log.Print("Added: ", body.Name)
_, err = config.Client.SendMessage(ctx, &sqs.SendMessageInput{
MessageAttributes: map[string]types.MessageAttributeValue{
"type": {
DataType: aws.String("String"),
StringValue: aws.String("CREATE_NAME_SUCCESS"),
},
},
QueueUrl: aws.String(config.URL),
MessageBody: aws.String("{}"),
})
if err != nil {
return err
}
_, err = config.Client.DeleteMessage(ctx, &sqs.DeleteMessageInput{
QueueUrl: aws.String(config.URL),
ReceiptHandle: msg.ReceiptHandle,
})
if err != nil {
return err
}
return nil
}
api/queue/poll.go
+47
View File
@@ -0,0 +1,47 @@
package queue
import (
"context"
"log"
"github.com/aws/aws-sdk-go-v2/service/sqs"
)
type Controllers struct {
Name NameController
}
type Queue struct {
Config *QueueConfig
Controllers *Controllers
}
func PollMessages(ctx context.Context, queue *Queue) {
for {
result, err := queue.Config.Client.ReceiveMessage(ctx, &sqs.ReceiveMessageInput{
QueueUrl: &queue.Config.URL,
MaxNumberOfMessages: 1,
WaitTimeSeconds: 2,
VisibilityTimeout: 2,
MessageAttributeNames: []string{
"type",
},
})
if err != nil {
log.Printf("Message Fetch Fail: %v", err)
continue
}
for _, message := range result.Messages {
go func() {
toProcess, err := processMessage(ctx, queue, message)
if !toProcess {
return
}
if err != nil {
log.Printf("Message Process Fail: %v", err)
}
}()
}
}
}
api/queue/process.go
+24
View File
@@ -0,0 +1,24 @@
package queue
import (
"context"
"github.com/aws/aws-sdk-go-v2/service/sqs"
"github.com/aws/aws-sdk-go-v2/service/sqs/types"
)
type QueueConfig struct {
URL string
Client *sqs.Client
}
func processMessage(ctx context.Context, queue *Queue, message types.Message) (bool, error) {
// Process the message here and return false if not message to process
switch *message.MessageAttributes["type"].StringValue {
case "CREATE_NAME":
err := queue.Controllers.Name.Create(ctx, queue.Config, &message)
return true, err
}
return false, nil
}