putting in base query and collector framework
This commit is contained in:
Michael McGuinness
@@ -1,45 +1,45 @@
|
||||
package database
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"gotemplate/internal/env"
|
||||
"log"
|
||||
|
||||
"github.com/jackc/pgx/v5"
|
||||
"github.com/jackc/pgx/v5/pgxpool"
|
||||
)
|
||||
|
||||
func createConnectionString() string {
|
||||
driver := "postgres"
|
||||
dbUser := env.GetRequired("DB_USER")
|
||||
dbPass := env.GetRequired("DB_PASS")
|
||||
dbHost := env.GetRequired("DB_HOST")
|
||||
dbPort := env.GetRequired("DB_PORT")
|
||||
dbName := env.GetRequired("DB_NAME")
|
||||
connStr := fmt.Sprintf("%s://%s:%s@%s:%s/%s?sslmode=disable", driver, dbUser, dbPass, dbHost, dbPort, dbName)
|
||||
|
||||
return connStr
|
||||
}
|
||||
|
||||
func GetDBPool(ctx context.Context) *pgxpool.Pool {
|
||||
connStr := createConnectionString()
|
||||
|
||||
pool, err := pgxpool.New(ctx, connStr)
|
||||
if err != nil {
|
||||
log.Fatalf("Unable to create database pool: %v\n", err)
|
||||
}
|
||||
|
||||
return pool
|
||||
}
|
||||
|
||||
func GetDBConn(ctx context.Context) *pgx.Conn {
|
||||
connStr := createConnectionString()
|
||||
|
||||
conn, err := pgx.Connect(ctx, connStr)
|
||||
if err != nil {
|
||||
log.Fatalf("Unable to connect to database: %v\n", err)
|
||||
}
|
||||
|
||||
return conn
|
||||
}
|
||||
package database
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"gotemplate/internal/env"
|
||||
"log"
|
||||
|
||||
"github.com/jackc/pgx/v5"
|
||||
"github.com/jackc/pgx/v5/pgxpool"
|
||||
)
|
||||
|
||||
func createConnectionString() string {
|
||||
driver := "postgres"
|
||||
dbUser := env.GetRequired("DB_USER")
|
||||
dbPass := env.GetRequired("DB_PASS")
|
||||
dbHost := env.GetRequired("DB_HOST")
|
||||
dbPort := env.GetRequired("DB_PORT")
|
||||
dbName := env.GetRequired("DB_NAME")
|
||||
connStr := fmt.Sprintf("%s://%s:%s@%s:%s/%s?sslmode=disable", driver, dbUser, dbPass, dbHost, dbPort, dbName)
|
||||
|
||||
return connStr
|
||||
}
|
||||
|
||||
func GetDBPool(ctx context.Context) *pgxpool.Pool {
|
||||
connStr := createConnectionString()
|
||||
|
||||
pool, err := pgxpool.New(ctx, connStr)
|
||||
if err != nil {
|
||||
log.Fatalf("Unable to create database pool: %v\n", err)
|
||||
}
|
||||
|
||||
return pool
|
||||
}
|
||||
|
||||
func GetDBConn(ctx context.Context) *pgx.Conn {
|
||||
connStr := createConnectionString()
|
||||
|
||||
conn, err := pgx.Connect(ctx, connStr)
|
||||
if err != nil {
|
||||
log.Fatalf("Unable to connect to database: %v\n", err)
|
||||
}
|
||||
|
||||
return conn
|
||||
}
|
||||
|
||||
@@ -1,85 +1,85 @@
|
||||
package database
|
||||
|
||||
import (
|
||||
"database/sql"
|
||||
"fmt"
|
||||
"gotemplate/internal/env"
|
||||
"log"
|
||||
"os"
|
||||
"path"
|
||||
|
||||
"github.com/golang-migrate/migrate/v4"
|
||||
"github.com/golang-migrate/migrate/v4/database/postgres"
|
||||
_ "github.com/golang-migrate/migrate/v4/source/file"
|
||||
)
|
||||
|
||||
func createDB() {
|
||||
dbUser := env.GetRequired("DB_USER")
|
||||
dbPass := env.GetRequired("DB_PASS")
|
||||
dbHost := env.GetRequired("DB_HOST")
|
||||
dbPort := env.GetRequired("DB_PORT")
|
||||
dbName := env.GetRequired("DB_NAME")
|
||||
driver := "postgres"
|
||||
connStr := fmt.Sprintf("%s://%s:%s@%s:%s?sslmode=disable", driver, dbUser, dbPass, dbHost, dbPort)
|
||||
|
||||
db, err := sql.Open(driver, connStr)
|
||||
if err != nil {
|
||||
log.Fatalf("Error opening database: %v", err)
|
||||
}
|
||||
defer db.Close()
|
||||
|
||||
result, err := db.Exec(fmt.Sprintf("SELECT 'CREATE DATABASE %s' WHERE NOT EXISTS (SELECT FROM pg_database WHERE datname = '%s')", dbName, dbName))
|
||||
if err != nil {
|
||||
log.Fatalf("Error creating database: %v", err)
|
||||
}
|
||||
|
||||
rows, err := result.RowsAffected()
|
||||
if err != nil {
|
||||
log.Fatal(err)
|
||||
} else if rows == 0 {
|
||||
log.Printf("Database already exists: %s", dbName)
|
||||
} else {
|
||||
log.Printf("Database created: %s", dbName)
|
||||
}
|
||||
}
|
||||
|
||||
func RunMigrations() {
|
||||
createDB()
|
||||
|
||||
connStr := createConnectionString()
|
||||
driver := "postgres"
|
||||
|
||||
db, err := sql.Open(driver, connStr)
|
||||
if err != nil {
|
||||
log.Fatalf("failed to open database: %v", err)
|
||||
}
|
||||
defer db.Close()
|
||||
|
||||
if err = db.Ping(); err != nil {
|
||||
log.Fatal("Failed to connect to database:", err)
|
||||
}
|
||||
|
||||
conn, err := postgres.WithInstance(db, &postgres.Config{})
|
||||
if err != nil {
|
||||
log.Fatal(err)
|
||||
}
|
||||
|
||||
migPath := "file://" + path.Join(os.Getenv("PWD"), "database/migrations")
|
||||
m, err := migrate.NewWithDatabaseInstance(
|
||||
migPath,
|
||||
driver,
|
||||
conn,
|
||||
)
|
||||
if err != nil {
|
||||
log.Fatalf("failed to create migrate instance: %v", err)
|
||||
}
|
||||
|
||||
err = m.Up()
|
||||
if err == migrate.ErrNoChange {
|
||||
log.Println("No changes required")
|
||||
} else if err != nil {
|
||||
log.Fatalf("failed to apply migrations: %v", err)
|
||||
} else {
|
||||
log.Println("Migrations applied successfully!")
|
||||
}
|
||||
}
|
||||
package database
|
||||
|
||||
import (
|
||||
"database/sql"
|
||||
"fmt"
|
||||
"gotemplate/internal/env"
|
||||
"log"
|
||||
"os"
|
||||
"path"
|
||||
|
||||
"github.com/golang-migrate/migrate/v4"
|
||||
"github.com/golang-migrate/migrate/v4/database/postgres"
|
||||
_ "github.com/golang-migrate/migrate/v4/source/file"
|
||||
)
|
||||
|
||||
func createDB() {
|
||||
dbUser := env.GetRequired("DB_USER")
|
||||
dbPass := env.GetRequired("DB_PASS")
|
||||
dbHost := env.GetRequired("DB_HOST")
|
||||
dbPort := env.GetRequired("DB_PORT")
|
||||
dbName := env.GetRequired("DB_NAME")
|
||||
driver := "postgres"
|
||||
connStr := fmt.Sprintf("%s://%s:%s@%s:%s?sslmode=disable", driver, dbUser, dbPass, dbHost, dbPort)
|
||||
|
||||
db, err := sql.Open(driver, connStr)
|
||||
if err != nil {
|
||||
log.Fatalf("Error opening database: %v", err)
|
||||
}
|
||||
defer db.Close()
|
||||
|
||||
result, err := db.Exec(fmt.Sprintf("SELECT 'CREATE DATABASE %s' WHERE NOT EXISTS (SELECT FROM pg_database WHERE datname = '%s')", dbName, dbName))
|
||||
if err != nil {
|
||||
log.Fatalf("Error creating database: %v", err)
|
||||
}
|
||||
|
||||
rows, err := result.RowsAffected()
|
||||
if err != nil {
|
||||
log.Fatal(err)
|
||||
} else if rows == 0 {
|
||||
log.Printf("Database already exists: %s", dbName)
|
||||
} else {
|
||||
log.Printf("Database created: %s", dbName)
|
||||
}
|
||||
}
|
||||
|
||||
func RunMigrations() {
|
||||
createDB()
|
||||
|
||||
connStr := createConnectionString()
|
||||
driver := "postgres"
|
||||
|
||||
db, err := sql.Open(driver, connStr)
|
||||
if err != nil {
|
||||
log.Fatalf("failed to open database: %v", err)
|
||||
}
|
||||
defer db.Close()
|
||||
|
||||
if err = db.Ping(); err != nil {
|
||||
log.Fatal("Failed to connect to database:", err)
|
||||
}
|
||||
|
||||
conn, err := postgres.WithInstance(db, &postgres.Config{})
|
||||
if err != nil {
|
||||
log.Fatal(err)
|
||||
}
|
||||
|
||||
migPath := "file://" + path.Join(os.Getenv("PWD"), "database/migrations")
|
||||
m, err := migrate.NewWithDatabaseInstance(
|
||||
migPath,
|
||||
driver,
|
||||
conn,
|
||||
)
|
||||
if err != nil {
|
||||
log.Fatalf("failed to create migrate instance: %v", err)
|
||||
}
|
||||
|
||||
err = m.Up()
|
||||
if err == migrate.ErrNoChange {
|
||||
log.Println("No changes required")
|
||||
} else if err != nil {
|
||||
log.Fatalf("failed to apply migrations: %v", err)
|
||||
} else {
|
||||
log.Println("Migrations applied successfully!")
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1,32 +1,32 @@
|
||||
// Code generated by sqlc. DO NOT EDIT.
|
||||
// versions:
|
||||
// sqlc v1.27.0
|
||||
|
||||
package repository
|
||||
|
||||
import (
|
||||
"context"
|
||||
|
||||
"github.com/jackc/pgx/v5"
|
||||
"github.com/jackc/pgx/v5/pgconn"
|
||||
)
|
||||
|
||||
type DBTX interface {
|
||||
Exec(context.Context, string, ...interface{}) (pgconn.CommandTag, error)
|
||||
Query(context.Context, string, ...interface{}) (pgx.Rows, error)
|
||||
QueryRow(context.Context, string, ...interface{}) pgx.Row
|
||||
}
|
||||
|
||||
func New(db DBTX) *Queries {
|
||||
return &Queries{db: db}
|
||||
}
|
||||
|
||||
type Queries struct {
|
||||
db DBTX
|
||||
}
|
||||
|
||||
func (q *Queries) WithTx(tx pgx.Tx) *Queries {
|
||||
return &Queries{
|
||||
db: tx,
|
||||
}
|
||||
}
|
||||
// Code generated by sqlc. DO NOT EDIT.
|
||||
// versions:
|
||||
// sqlc v1.27.0
|
||||
|
||||
package repository
|
||||
|
||||
import (
|
||||
"context"
|
||||
|
||||
"github.com/jackc/pgx/v5"
|
||||
"github.com/jackc/pgx/v5/pgconn"
|
||||
)
|
||||
|
||||
type DBTX interface {
|
||||
Exec(context.Context, string, ...interface{}) (pgconn.CommandTag, error)
|
||||
Query(context.Context, string, ...interface{}) (pgx.Rows, error)
|
||||
QueryRow(context.Context, string, ...interface{}) pgx.Row
|
||||
}
|
||||
|
||||
func New(db DBTX) *Queries {
|
||||
return &Queries{db: db}
|
||||
}
|
||||
|
||||
type Queries struct {
|
||||
db DBTX
|
||||
}
|
||||
|
||||
func (q *Queries) WithTx(tx pgx.Tx) *Queries {
|
||||
return &Queries{
|
||||
db: tx,
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1,9 +1,9 @@
|
||||
// Code generated by sqlc. DO NOT EDIT.
|
||||
// versions:
|
||||
// sqlc v1.27.0
|
||||
|
||||
package repository
|
||||
|
||||
type Name struct {
|
||||
Name string
|
||||
}
|
||||
// Code generated by sqlc. DO NOT EDIT.
|
||||
// versions:
|
||||
// sqlc v1.27.0
|
||||
|
||||
package repository
|
||||
|
||||
type Name struct {
|
||||
Name string
|
||||
}
|
||||
|
||||
@@ -1,30 +1,30 @@
|
||||
// Code generated by sqlc. DO NOT EDIT.
|
||||
// versions:
|
||||
// sqlc v1.27.0
|
||||
// source: name.sql
|
||||
|
||||
package repository
|
||||
|
||||
import (
|
||||
"context"
|
||||
)
|
||||
|
||||
const addName = `-- name: AddName :exec
|
||||
INSERT INTO names (name) VALUES ($1)
|
||||
`
|
||||
|
||||
func (q *Queries) AddName(ctx context.Context, name string) error {
|
||||
_, err := q.db.Exec(ctx, addName, name)
|
||||
return err
|
||||
}
|
||||
|
||||
const getName = `-- name: GetName :one
|
||||
SELECT name as dbName FROM names where name = $1
|
||||
`
|
||||
|
||||
func (q *Queries) GetName(ctx context.Context, name string) (string, error) {
|
||||
row := q.db.QueryRow(ctx, getName, name)
|
||||
var dbname string
|
||||
err := row.Scan(&dbname)
|
||||
return dbname, err
|
||||
}
|
||||
// Code generated by sqlc. DO NOT EDIT.
|
||||
// versions:
|
||||
// sqlc v1.27.0
|
||||
// source: name.sql
|
||||
|
||||
package repository
|
||||
|
||||
import (
|
||||
"context"
|
||||
)
|
||||
|
||||
const addName = `-- name: AddName :exec
|
||||
INSERT INTO names (name) VALUES ($1)
|
||||
`
|
||||
|
||||
func (q *Queries) AddName(ctx context.Context, name string) error {
|
||||
_, err := q.db.Exec(ctx, addName, name)
|
||||
return err
|
||||
}
|
||||
|
||||
const getName = `-- name: GetName :one
|
||||
SELECT name as dbName FROM names where name = $1
|
||||
`
|
||||
|
||||
func (q *Queries) GetName(ctx context.Context, name string) (string, error) {
|
||||
row := q.db.QueryRow(ctx, getName, name)
|
||||
var dbname string
|
||||
err := row.Scan(&dbname)
|
||||
return dbname, err
|
||||
}
|
||||
|
||||
@@ -0,0 +1,30 @@
|
||||
package document
|
||||
|
||||
import (
|
||||
"context"
|
||||
"gotemplate/internal/database/repository"
|
||||
)
|
||||
|
||||
type Service struct {
|
||||
db *repository.Queries
|
||||
}
|
||||
|
||||
type Document struct {
|
||||
ID string `json:"id"`
|
||||
JobID string `json:"jobId"`
|
||||
Name string `json:"name"`
|
||||
}
|
||||
|
||||
func New(ctx context.Context, db *repository.Queries) *Service {
|
||||
return &Service{
|
||||
db: db,
|
||||
}
|
||||
}
|
||||
|
||||
func (s *Service) Sync(ctx context.Context, doc Document) error {
|
||||
// Check if collector exists for doc and is synced
|
||||
// Check if job set up
|
||||
// If not find all outputs and see if any are out of sync/missing
|
||||
// create dependency tree - if sync triggered, sync all dependent outputs by using queue
|
||||
return nil
|
||||
}
|
||||
Vendored
+24
-24
@@ -1,24 +1,24 @@
|
||||
package env
|
||||
|
||||
import (
|
||||
"log"
|
||||
"os"
|
||||
)
|
||||
|
||||
func GetRequired(name string) string {
|
||||
value, exists := os.LookupEnv(name)
|
||||
if !exists {
|
||||
log.Fatal("Environment variable not set: ", name)
|
||||
}
|
||||
|
||||
return value
|
||||
}
|
||||
|
||||
func Get(name string) string {
|
||||
value, exists := os.LookupEnv(name)
|
||||
if !exists {
|
||||
log.Print("Environment variable not set: ", name)
|
||||
}
|
||||
|
||||
return value
|
||||
}
|
||||
package env
|
||||
|
||||
import (
|
||||
"log"
|
||||
"os"
|
||||
)
|
||||
|
||||
func GetRequired(name string) string {
|
||||
value, exists := os.LookupEnv(name)
|
||||
if !exists {
|
||||
log.Fatal("Environment variable not set: ", name)
|
||||
}
|
||||
|
||||
return value
|
||||
}
|
||||
|
||||
func Get(name string) string {
|
||||
value, exists := os.LookupEnv(name)
|
||||
if !exists {
|
||||
log.Print("Environment variable not set: ", name)
|
||||
}
|
||||
|
||||
return value
|
||||
}
|
||||
|
||||
@@ -1,24 +0,0 @@
|
||||
package name
|
||||
|
||||
import (
|
||||
"context"
|
||||
"gotemplate/internal/database/repository"
|
||||
)
|
||||
|
||||
type Service struct {
|
||||
db *repository.Queries
|
||||
}
|
||||
|
||||
func New(ctx context.Context, db *repository.Queries) *Service {
|
||||
return &Service{
|
||||
db: db,
|
||||
}
|
||||
}
|
||||
|
||||
func (s *Service) Add(ctx context.Context, name string) error {
|
||||
return s.db.AddName(ctx, name)
|
||||
}
|
||||
|
||||
func (s *Service) Get(ctx context.Context, name string) (string, error) {
|
||||
return s.db.GetName(ctx, name)
|
||||
}
|
||||
+33
-33
@@ -1,33 +1,33 @@
|
||||
package otel
|
||||
|
||||
import (
|
||||
"context"
|
||||
"log"
|
||||
"os"
|
||||
|
||||
"go.opentelemetry.io/otel"
|
||||
"go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc"
|
||||
sdktrace "go.opentelemetry.io/otel/sdk/trace"
|
||||
)
|
||||
|
||||
func Init(ctx context.Context) func() {
|
||||
if os.Getenv("ENABLE_OTEL") != "1" {
|
||||
log.Println("OpenTelemetry is disabled. Set ENABLE_OTEL to enable.")
|
||||
return func() {} // No-op shutdown function
|
||||
}
|
||||
|
||||
exporter, err := otlptracegrpc.New(ctx)
|
||||
if err != nil {
|
||||
log.Fatal(err)
|
||||
}
|
||||
|
||||
tp := sdktrace.NewTracerProvider(
|
||||
sdktrace.WithBatcher(exporter),
|
||||
)
|
||||
|
||||
otel.SetTracerProvider(tp)
|
||||
|
||||
return func() {
|
||||
_ = tp.Shutdown(ctx)
|
||||
}
|
||||
}
|
||||
package otel
|
||||
|
||||
import (
|
||||
"context"
|
||||
"log"
|
||||
"os"
|
||||
|
||||
"go.opentelemetry.io/otel"
|
||||
"go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc"
|
||||
sdktrace "go.opentelemetry.io/otel/sdk/trace"
|
||||
)
|
||||
|
||||
func Init(ctx context.Context) func() {
|
||||
if os.Getenv("ENABLE_OTEL") != "1" {
|
||||
log.Println("OpenTelemetry is disabled. Set ENABLE_OTEL to enable.")
|
||||
return func() {} // No-op shutdown function
|
||||
}
|
||||
|
||||
exporter, err := otlptracegrpc.New(ctx)
|
||||
if err != nil {
|
||||
log.Fatal(err)
|
||||
}
|
||||
|
||||
tp := sdktrace.NewTracerProvider(
|
||||
sdktrace.WithBatcher(exporter),
|
||||
)
|
||||
|
||||
otel.SetTracerProvider(tp)
|
||||
|
||||
return func() {
|
||||
_ = tp.Shutdown(ctx)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,10 @@
|
||||
package queue
|
||||
|
||||
import (
|
||||
"github.com/aws/aws-sdk-go-v2/service/sqs"
|
||||
)
|
||||
|
||||
type QueueConfig struct {
|
||||
URL string
|
||||
Client *sqs.Client
|
||||
}
|
||||
@@ -0,0 +1,21 @@
|
||||
package queue
|
||||
|
||||
import (
|
||||
"context"
|
||||
|
||||
"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"
|
||||
)
|
||||
|
||||
func Delete(ctx context.Context, config *QueueConfig, msg *types.Message) error {
|
||||
_, err := config.Client.DeleteMessage(ctx, &sqs.DeleteMessageInput{
|
||||
QueueUrl: aws.String(config.URL),
|
||||
ReceiptHandle: msg.ReceiptHandle,
|
||||
})
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
@@ -0,0 +1,35 @@
|
||||
package queue
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
|
||||
"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"
|
||||
)
|
||||
|
||||
func Send(ctx context.Context, config *QueueConfig, typeName string, body interface{}) error {
|
||||
jsonBytes, err := json.Marshal(body)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
strBody := string(jsonBytes)
|
||||
|
||||
_, err = config.Client.SendMessage(ctx, &sqs.SendMessageInput{
|
||||
MessageAttributes: map[string]types.MessageAttributeValue{
|
||||
"type": {
|
||||
DataType: aws.String("String"),
|
||||
StringValue: aws.String(typeName),
|
||||
},
|
||||
},
|
||||
QueueUrl: aws.String(config.URL),
|
||||
MessageBody: aws.String(strBody),
|
||||
})
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
Reference in New Issue
Block a user