feat: completion of tasks to do

.idea/misc.xml

@@ -1,6 +1,6 @@
 <?xml version="1.0" encoding="UTF-8"?>
 <project version="4">
-  <component name="ProjectRootManager">
+  <component name="ProjectRootManager" version="2" languageLevel="JDK_20" project-jdk-name="Python 3.11 (transaction-watcher)" project-jdk-type="Python SDK">
     <output url="file://$PROJECT_DIR$/out" />
   </component>
 </project>

.idea/transaction-watcher.iml

@@ -1,12 +1,18 @@
 <?xml version="1.0" encoding="UTF-8"?>
 <module type="JAVA_MODULE" version="4">
+  <component name="Go" enabled="true" />
   <component name="NewModuleRootManager" inherit-compiler-output="true">
     <exclude-output />
     <content url="file://$MODULE_DIR$">
       <sourceFolder url="file://$MODULE_DIR$/inserter/src" isTestSource="false" />
       <excludeFolder url="file://$MODULE_DIR$/inserter/target" />
+      <excludeFolder url="file://$MODULE_DIR$/venv" />
     </content>
     <orderEntry type="inheritedJdk" />
     <orderEntry type="sourceFolder" forTests="false" />
   </component>
+  <component name="PackageRequirementsSettings">
+    <option name="requirementsPath" value="C:\Users\reina\Repositories\personal\transaction-watcher\customer-list\requirements.txt" />
+    <option name="removeUnused" value="true" />
+  </component>
 </module>

.idea/vcs.xml

@@ -2,6 +2,5 @@
 <project version="4">
   <component name="VcsDirectoryMappings">
     <mapping directory="$PROJECT_DIR$" vcs="Git" />
-    <mapping directory="$PROJECT_DIR$/inserter" vcs="Git" />
   </component>
 </project>

README.md

@@ -1 +1,103 @@
-# Transaction Watcher
+# Transaction Watcher
+
+This is a repository to harden your SQL skill, learn Kafka (or Redpanda) and learn Docker.
+
+## Your Tasks
+
+This project contains a few tasks for you to work with.
+
+### #1 The Watcher
+
+Create a subdirectory called `watcher` that will do a SQL query periodically to the `transactions` table, 
+and for every new row on that table, the `watcher` will produce a message to `transactions` topic on Redpanda.
+
+The schema of the message is defined on the `kafka-schemas` directory. Search for `transaction.json` file.
+
+You must create a Dockerfile for your application. You can choose any language.
+
+You don't need to create a HTTP API for this one. Just create a single function that will run when the 
+program is executed. The application *must* work without any interference from you. It must work without
+having anyone (including you) to trigger the run or consume function.
+
+You can see that your `watcher` is emitting correct message to Redpanda through Redpanda Console that is
+running on your local machine on port `8080`.
+
+There will be environment variable available for you when you run it though Docker Compose:
+
+```yaml
+DATABASE_URL: "postgresql://watcher:password@postgres:5432/watcher?sslmode=disable"
+KAFKA_ADDRESSES: "kafka:9092"
+```
+
+### #2 The Swimmer
+
+Create a subdirectory called `swimmer` that will consume the `balance` topic from Redpanda, and do an event sourcing
+of a customer's current balance. You must expose a HTTP API that handles a single endpoint of:
+
+```http request
+GET /current-balance?customer_id=123
+Accept: application/json
+```
+
+With a response schema of:
+```json
+{
+  "customer_id": 123,
+  "current_balance": 123456
+}
+```
+
+It's called swimmer because you will swim through the Redpanda `balance` topic records, and get an answer from that.
+I will not give a clue on how you should consume your topic and produce a result based on the customer ID.
+
+For the `balance` topic schema, you can look for `balance.json` on `kafka-schemas` directory.
+
+You must create a Dockerfile for your application. You can choose any language.
+
+Please expose the HTTP API at port 3000.
+
+There will be environment variable available for you when you run it though Docker Compose:
+
+```yaml
+KAFKA_ADDRESSES: "kafka:9092"
+```
+
+### #3 The Frontend
+
+This is an optional task.
+
+Create a frontend (on `frontend` directory) that shows list of customer ID from `customer-list` service. 
+Then, show the amount or balance that each customer have. You can create a full SPA page, or an SSR website.
+
+You can hit these two endpoints:
+
+```http request
+GET http://customer-list:7201/customers
+Accept: application/json
+
+# Returns list of customer ID
+[1,2,3,4,5]
+```
+
+```http request
+GET http://swimmer:3000/current-balance?customer_id=123
+Accept: application/json
+
+# Returns:
+{
+  "customer_id": 123,
+  "current_balance": 123456
+}
+```
+
+Hey, the `swimmer` one is the one you made!
+
+If you are developing it from your local machine, you can replace both `customer-list` and `swimmer` hostname
+into `localhost`.
+
+You can create a Dockerfile and add a new service schema on `docker-compose.yml` file to state your frontend container
+configuration.
+
+## Words of affirmation
+
+Good luck, you can do it!

balance-processor/.gitignore

@@ -0,0 +1,23 @@
+### Go template
+# If you prefer the allow list template instead of the deny list, see community template:
+# https://github.com/github/gitignore/blob/main/community/Golang/Go.AllowList.gitignore
+#
+# Binaries for programs and plugins
+*.exe
+*.exe~
+*.dll
+*.so
+*.dylib
+
+# Test binary, built with `go test -c`
+*.test
+
+# Output of the go coverage tool, specifically when used with LiteIDE
+*.out
+
+# Dependency directories (remove the comment below to include it)
+# vendor/
+
+# Go workspace file
+go.work
+

balance-processor/Dockerfile

@@ -0,0 +1,11 @@
+FROM golang:1.20.3-bullseye AS builder
+WORKDIR /app
+COPY . .
+RUN go build -o balance-processor .
+
+FROM debian:bullseye AS runtime
+WORKDIR /app
+RUN apt-get update && apt-get install -y curl
+COPY --from=builder /app/balance-processor balance-processor
+
+CMD ["/app/balance-processor"]

balance-processor/consume.go

@@ -0,0 +1,47 @@
+package main
+
+import (
+	"encoding/json"
+	"fmt"
+	"github.com/twmb/franz-go/pkg/kgo"
+	"strings"
+	"time"
+)
+
+type TransactionLog struct {
+	TransactionID     int64  `json:"transaction_id"`
+	TransactionType   string `json:"transaction_type"`
+	CustomerNumber    int64  `json:"customer_number"`
+	TransactionAmount int64  `json:"transaction_amount"`
+	Timestamp         string `json:"timestamp"`
+}
+
+func consume(client *kgo.Client, payload []byte) error {
+	var transactionLog TransactionLog
+	if err := json.Unmarshal(payload, &transactionLog); err != nil {
+		return fmt.Errorf("invalid payload: %s", err.Error())
+	}
+
+	var amount int64
+	switch strings.ToUpper(transactionLog.TransactionType) {
+	case "TOP_UP":
+		amount = transactionLog.TransactionAmount
+		break
+	case "TRANSFER":
+		fallthrough
+	case "WITHDRAW":
+		fallthrough
+	case "FEE":
+		amount = -1 * transactionLog.TransactionAmount
+		break
+	default:
+		return fmt.Errorf("unknown transaction type: %s", transactionLog.TransactionType)
+	}
+
+	return produce(client, BalanceLog{
+		CustomerNumber: transactionLog.CustomerNumber,
+		Amount:         amount,
+		Timestamp:      time.Now().Format(time.RFC3339),
+		TransactionID:  transactionLog.TransactionID,
+	})
+}

balance-processor/go.mod

@@ -0,0 +1,11 @@
+module balance-processor
+
+go 1.20
+
+require github.com/twmb/franz-go v1.13.2
+
+require (
+	github.com/klauspost/compress v1.16.3 // indirect
+	github.com/pierrec/lz4/v4 v4.1.17 // indirect
+	github.com/twmb/franz-go/pkg/kmsg v1.4.0 // indirect
+)

balance-processor/go.sum

@@ -0,0 +1,8 @@
+github.com/klauspost/compress v1.16.3 h1:XuJt9zzcnaz6a16/OU53ZjWp/v7/42WcR5t2a0PcNQY=
+github.com/klauspost/compress v1.16.3/go.mod h1:ntbaceVETuRiXiv4DpjP66DpAtAGkEQskQzEyD//IeE=
+github.com/pierrec/lz4/v4 v4.1.17 h1:kV4Ip+/hUBC+8T6+2EgburRtkE9ef4nbY3f4dFhGjMc=
+github.com/pierrec/lz4/v4 v4.1.17/go.mod h1:gZWDp/Ze/IJXGXf23ltt2EXimqmTUXEy0GFuRQyBid4=
+github.com/twmb/franz-go v1.13.2 h1:jIdDoFiq8uP3Zrx6TZZTXpaXrv3bh1w3tV5mn/B+Gw8=
+github.com/twmb/franz-go v1.13.2/go.mod h1:jm/FtYxmhxDTN0gNSb26XaJY0irdSVcsckLiR5tQNMk=
+github.com/twmb/franz-go/pkg/kmsg v1.4.0 h1:tbp9hxU6m8qZhQTlpGiaIJOm4BXix5lsuEZ7K00dF0s=
+github.com/twmb/franz-go/pkg/kmsg v1.4.0/go.mod h1:SxG/xJKhgPu25SamAq0rrucfp7lbzCpEXOC+vH/ELrY=

balance-processor/main.go

@@ -0,0 +1,69 @@
+package main
+
+import (
+	"context"
+	"errors"
+	"github.com/twmb/franz-go/pkg/kgo"
+	"github.com/twmb/franz-go/pkg/kversion"
+	"log"
+	"os"
+	"os/signal"
+	"strings"
+)
+
+func main() {
+	kafkaHost, ok := os.LookupEnv("KAFKA_HOST")
+	if !ok {
+		kafkaHost = "localhost:9092"
+	}
+
+	kafkaOptions := []kgo.Opt{
+		kgo.MinVersions(kversion.V0_11_0()),
+		kgo.SeedBrokers(strings.Split(kafkaHost, ",")...),
+		kgo.ConsumeTopics("transactions"),
+	}
+
+	kafkaClient, err := kgo.NewClient(kafkaOptions...)
+	if err != nil {
+		log.Fatalf("initiating kafka client: %s", err.Error())
+	}
+	defer kafkaClient.Close()
+
+	exitSignal := make(chan os.Signal, 1)
+	signal.Notify(exitSignal, os.Interrupt)
+
+	go func() {
+		<-exitSignal
+
+		kafkaClient.Close()
+	}()
+
+	log.Println("Ready, waiting for messages...")
+
+	for {
+		ctx := context.Background()
+		fetches := kafkaClient.PollRecords(ctx, 128)
+
+		switch {
+		case fetches.IsClientClosed():
+			break
+		case fetches.Err() != nil:
+			fetches.EachError(func(topic string, partition int32, err error) {
+				if !errors.Is(err, context.DeadlineExceeded) && !errors.Is(err, context.Canceled) {
+					log.Printf("Error: topic: %s, partition: %d, message: %v", topic, partition, err)
+				}
+			})
+			continue
+		}
+
+		for _, msg := range fetches.Records() {
+			err := consume(kafkaClient, msg.Value)
+			if err != nil {
+				log.Println(err)
+				continue
+			}
+
+			log.Printf("Processed: %s", string(msg.Value))
+		}
+	}
+}

balance-processor/produce.go

@@ -0,0 +1,39 @@
+package main
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"github.com/twmb/franz-go/pkg/kgo"
+	"log"
+	"time"
+)
+
+type BalanceLog struct {
+	CustomerNumber int64  `json:"customer_number"`
+	Amount         int64  `json:"amount"`
+	Timestamp      string `json:"timestamp"`
+	TransactionID  int64  `json:"transaction_id"`
+}
+
+func produce(client *kgo.Client, balanceLog BalanceLog) error {
+	payload, err := json.Marshal(balanceLog)
+	if err != nil {
+		return fmt.Errorf("marshaling json: %w", err)
+	}
+
+	ctx := context.Background()
+	client.Produce(ctx, &kgo.Record{
+		Value:     payload,
+		Headers:   nil,
+		Timestamp: time.Now(),
+		Topic:     "balance",
+		Context:   ctx,
+	}, func(_ *kgo.Record, err error) {
+		if err != nil {
+			log.Println(err)
+		}
+	})
+
+	return nil
+}

customer-list/.gitignore

@@ -0,0 +1,162 @@
+### Python template
+# Byte-compiled / optimized / DLL files
+__pycache__/
+*.py[cod]
+*$py.class
+
+# C extensions
+*.so
+
+# Distribution / packaging
+.Python
+build/
+develop-eggs/
+dist/
+downloads/
+eggs/
+.eggs/
+lib/
+lib64/
+parts/
+sdist/
+var/
+wheels/
+share/python-wheels/
+*.egg-info/
+.installed.cfg
+*.egg
+MANIFEST
+
+# PyInstaller
+#  Usually these files are written by a python script from a template
+#  before PyInstaller builds the exe, so as to inject date/other infos into it.
+*.manifest
+*.spec
+
+# Installer logs
+pip-log.txt
+pip-delete-this-directory.txt
+
+# Unit test / coverage reports
+htmlcov/
+.tox/
+.nox/
+.coverage
+.coverage.*
+.cache
+nosetests.xml
+coverage.xml
+*.cover
+*.py,cover
+.hypothesis/
+.pytest_cache/
+cover/
+
+# Translations
+*.mo
+*.pot
+
+# Django stuff:
+*.log
+local_settings.py
+db.sqlite3
+db.sqlite3-journal
+
+# Flask stuff:
+instance/
+.webassets-cache
+
+# Scrapy stuff:
+.scrapy
+
+# Sphinx documentation
+docs/_build/
+
+# PyBuilder
+.pybuilder/
+target/
+
+# Jupyter Notebook
+.ipynb_checkpoints
+
+# IPython
+profile_default/
+ipython_config.py
+
+# pyenv
+#   For a library or package, you might want to ignore these files since the code is
+#   intended to run in multiple environments; otherwise, check them in:
+# .python-version
+
+# pipenv
+#   According to pypa/pipenv#598, it is recommended to include Pipfile.lock in version control.
+#   However, in case of collaboration, if having platform-specific dependencies or dependencies
+#   having no cross-platform support, pipenv may install dependencies that don't work, or not
+#   install all needed dependencies.
+#Pipfile.lock
+
+# poetry
+#   Similar to Pipfile.lock, it is generally recommended to include poetry.lock in version control.
+#   This is especially recommended for binary packages to ensure reproducibility, and is more
+#   commonly ignored for libraries.
+#   https://python-poetry.org/docs/basic-usage/#commit-your-poetrylock-file-to-version-control
+#poetry.lock
+
+# pdm
+#   Similar to Pipfile.lock, it is generally recommended to include pdm.lock in version control.
+#pdm.lock
+#   pdm stores project-wide configurations in .pdm.toml, but it is recommended to not include it
+#   in version control.
+#   https://pdm.fming.dev/#use-with-ide
+.pdm.toml
+
+# PEP 582; used by e.g. github.com/David-OConnor/pyflow and github.com/pdm-project/pdm
+__pypackages__/
+
+# Celery stuff
+celerybeat-schedule
+celerybeat.pid
+
+# SageMath parsed files
+*.sage.py
+
+# Environments
+.env
+.venv
+env/
+venv/
+ENV/
+env.bak/
+venv.bak/
+
+# Spyder project settings
+.spyderproject
+.spyproject
+
+# Rope project settings
+.ropeproject
+
+# mkdocs documentation
+/site
+
+# mypy
+.mypy_cache/
+.dmypy.json
+dmypy.json
+
+# Pyre type checker
+.pyre/
+
+# pytype static type analyzer
+.pytype/
+
+# Cython debug symbols
+cython_debug/
+
+# PyCharm
+#  JetBrains specific template is maintained in a separate JetBrains.gitignore that can
+#  be found at https://github.com/github/gitignore/blob/main/Global/JetBrains.gitignore
+#  and can be added to the global gitignore or merged into this file.  For a more nuclear
+#  option (not recommended) you can uncomment the following to ignore the entire idea folder.
+#.idea/
+

customer-list/Dockerfile

@@ -0,0 +1,7 @@
+FROM python:3.11-bullseye
+WORKDIR /app
+RUN apt-get update && apt-get install -y curl python3-dev libpq-dev libpq5
+COPY . .
+RUN pip install -r requirements.txt
+
+CMD ["python3", "main.py"]

customer-list/main.py

@@ -0,0 +1,27 @@
+from flask import Flask
+import psycopg
+from waitress import serve
+import os
+from querier import querier
+
+database_url = os.environ['DATABASE_URL'] \
+    if os.environ['DATABASE_URL'] is not None and os.environ['DATABASE_URL'] != '' \
+    else 'postgresql://watcher:password@localhost:5432/watcher?sslmode=disable'
+
+app = Flask(__name__)
+
+with psycopg.connect(database_url) as conn:
+    conn.read_only = True
+
+    @app.get('/')
+    def hello():
+        return 'OK'
+
+
+    @app.get('/customers')
+    def customers():
+        return querier(conn)
+
+    serve(app, host="0.0.0.0", port=7201)
+
+

customer-list/querier.py

@@ -0,0 +1,13 @@
+from psycopg import Connection
+
+
+def querier(conn: Connection) -> list[str]:
+    customer_list: list[str] = []
+    with conn.cursor() as cursor:
+        cursor.execute("SELECT DISTINCT customer_number FROM transactions")
+        for record in cursor:
+            customer_list.append(*record)
+
+        conn.commit()
+
+    return customer_list

customer-list/requirements.txt

@@ -0,0 +1,7 @@
+pip~=22.3.1
+wheel~=0.38.4
+setuptools~=65.5.1
+Jinja2~=3.1.2
+psycopg~=3.1.8
+Flask~=2.2.3
+waitress~=2.1.2

docker-compose.yml

@@ -14,6 +14,8 @@ services:
       timeout: 10s
       retries: 10
       start_period: 30s
+    volumes:
+      - ./postgresql:/docker-entrypoint-initdb.d:ro
 
   kafka:
     image: docker.redpanda.com/vectorized/redpanda:v22.2.2
@@ -41,7 +43,7 @@ services:
       retries: 3
       start_period: 60s
 
-  # This is for vieweing the Kafka topics and the contents within it.
+  # This is for viewing the Kafka topics and the contents within it.
   # If you're like me, you don't really need this bit, just do unit test like usual.
   # If it's works, that it'll work.
   kafka-console:
@@ -61,4 +63,48 @@ services:
     restart: on-failure:5
     depends_on:
       postgres:
-        condition: service_healthy
+        condition: service_healthy
+
+  balance-processor:
+    build: ./balance-processor
+    environment:
+      KAFKA_ADDRESSES: "kafka:9092"
+    restart: on-failure:5
+    depends_on:
+      kafka:
+        condition: service_healthy
+
+  customer-list:
+    build: ./customer-list
+    ports:
+      - 7201:7201
+    environment:
+      DATABASE_URL: "postgresql://watcher:password@postgres:5432/watcher?sslmode=disable"
+    restart: on-failure:5
+    depends_on:
+      postgres:
+        condition: service_healthy
+
+  # This is your space now.
+  # If you are developing the application, you can comment these out.
+  watcher:
+    build: ./watcher
+    environment:
+      DATABASE_URL: "postgresql://watcher:password@postgres:5432/watcher?sslmode=disable"
+      KAFKA_ADDRESSES: "kafka:9092"
+    restart: on-failure:5
+    depends_on:
+      kafka:
+        condition: service_healthy
+      postgres:
+        condition: service_healthy
+
+  swimmer:
+    build: ./swimmer
+    environment:
+      KAFKA_ADDRESSES: "kafka:9092"
+    restart: on-failure:5
+    depends_on:
+      kafka:
+        condition: service_healthy
+

inserter/src/main.rs

@@ -4,11 +4,12 @@ use std::sync::Arc;
 use std::thread;
 use std::time;
 use rand::{thread_rng, Rng};
-use postgres::{Client, Error, NoTls, Transaction};
+use postgres::{Client, NoTls};
 
 fn main() {
-    let database_url = env::var("DATABASE_URL").unwrap_or(String::from("postgresql://watcher:password@localhost:5432/watcher?sslmode=disable"));
-    let mut client = Client::connect(database_url.as_str(), NoTls)?;
+    let database_url = env::var("DATABASE_URL")
+        .unwrap_or(String::from("postgresql://watcher:password@localhost:5432/watcher?sslmode=disable"));
+    let mut client = Client::connect(database_url.as_str(), NoTls).unwrap();
     let mut rng = thread_rng();
 
     let running = Arc::new(AtomicBool::new(true));
@@ -16,11 +17,10 @@ fn main() {
 
     ctrlc::set_handler(move || {
         r.store(false, Ordering::SeqCst);
-        client.close()?;
     }).expect("Error setting Ctrl-C handler");
 
-    while running.load(Ordering::SeqCst)  {
-        let transaction_amount = rng.gen_range(-1e5..=1e9);
+    while running.load(Ordering::SeqCst) {
+        let transaction_amount: i64 = rng.gen_range(0..1e9 as i64);
         let transaction_type = match rng.gen_range(0..=3) {
             0 => String::from("TOP_UP"),
             1 => String::from("TRANSFER"),
@@ -30,37 +30,28 @@ fn main() {
         };
         let customer_number = rng.gen_range(100..200);
 
-        match client.transaction() {
-            Ok(mut transaction) => {
-                match transaction.execute(
-                    "INSERT INTO transactions (transaction_type, customer_number, transaction_amount, timestamp) VALUES ($1, $2, $3, NOW())",
-                    &[transaction_type, customer_number, transaction_amount]
-                ) {
-                    Ok(_) => {}
-                    Err(e1) => {
-                        eprintln!("{:?}", e1);
-                        if let Err(e2) = transaction.rollback() {
-                            eprintln!("{:?}", e2)
-                        }
-                    }
-                }
+        let exec_error = client.execute(
+            "INSERT INTO transactions (transaction_type, customer_number, transaction_amount, timestamp) VALUES ($1, $2, $3, NOW())",
+            &[&transaction_type, &customer_number, &transaction_amount]
+        ).err();
 
-                match transaction.commit() {
-                    Ok(_) => {}
-                    Err(e1) => {
-                        eprintln!("{:?}", e1);
-                        if let Err(e2) = transaction.rollback() {
-                            eprintln!("{:?}", e2)
-                        }
-                    }
-                }
-            }
-            Err(error) => {
-                eprintf!("{:?}", error)
+        println!("{}", format!(
+                "INSERT INTO transactions (transaction_type, customer_number, transaction_amount, timestamp) VALUES ({0}, {1}, {2}, NOW())",
+                transaction_type,
+                customer_number,
+                transaction_amount
+            ).to_string());
+
+        match exec_error {
+            None => {}
+            Some(e) => {
+                println!("{:?}", e);
             }
         }
 
         let time_to_sleep = time::Duration::from_millis(rng.gen_range(10..1000));
         thread::sleep(time_to_sleep);
     }
+
+    client.close().unwrap();
 }

kafka-schemas/balance.json

@@ -0,0 +1,6 @@
+{
+  "customer_number": 123456,
+  "amount": -123456,
+  "timestamp": "RFC3339 or ISO8601 FORMAT",
+  "transaction_id": "STRING"
+}

kafka-schemas/transactions.json

@@ -0,0 +1,7 @@
+{
+  "transaction_id": 123456,
+  "transaction_type": "STRING",
+  "customer_number": 123456,
+  "transaction_amount": 123456,
+  "timestamp": "RFC3339 or ISO8601 FORMAT"
+}

swimmer/Dockerfile

@@ -0,0 +1 @@
+FROM stratch

watcher/Dockerfile

@@ -0,0 +1 @@
+FROM stratch