完整源码
这里列出下载包中的全部文件，包括业务代码、存储、协议入口和测试。教程正文的代码节选也从同一份工程提取。
.gitignore
*.db
*.db-shm
*.db-wal
*.exeREADME.md
# KPro orders — 教学工程
配套教程：官网 `learn.html`。这是可以独立运行的 Go module，固定使用
KPro **v0.2.1**，没有指向开发者电脑的 `replace`。
需要 Go **1.26.9 或更新的兼容版本**（KPro v0.2.1 的要求）。
第一次 `go mod download` 需要访问 Go 模块下载源。SQLite 使用纯 Go 驱动，
不需要安装数据库服务器或 C 编译器。`-race` 检查另有平台工具链要求。
## 先运行成品
在本目录打开终端：
```sh
go mod download
go run ./cmd/orders
```
另开一个终端，仍在本目录：
```sh
go run ./cmd/check
```
检查工具会实际创建订单，查询、发送非法数量、取消并重复取消，打印每个 HTTP
状态码及 JSON。默认只监听本机 `127.0.0.1:8080`，不要直接暴露到公网。
订单存到当前工作目录的 `orders.db`。Ctrl+C 停止服务后重新启动，原订单仍可查询。
也可以手动请求（Windows PowerShell 请将 `curl` 写成 `curl.exe`）：
```sh
curl -i -H "Content-Type: application/json" --data-binary "@requests/create.json" http://127.0.0.1:8080/orders
```
记下返回的 `id`，将下面的 `ORDER_ID` 换成它：
```sh
curl -i http://127.0.0.1:8080/orders/ORDER_ID
curl -i -X POST http://127.0.0.1:8080/orders/ORDER_ID/cancel
```
## 三个可运行的观察点
各命令默认占用同一个 HTTP 端口；切换前先 Ctrl+C 停止上一个进程。
```sh
# 只有一个 /hello 路由：完整源码在 cmd/hello/main.go。
go run ./cmd/hello
# 完整业务，暂时不持久化；重启后数据消失。
go run ./cmd/orders -memory
# SQLite 持久化，并打开 TCP 查询入口。
go run ./cmd/orders -tcp 127.0.0.1:8088
```
TCP 客户端读取同一个服务内、由 HTTP 创建的订单：
```sh
go run ./cmd/ordercli -id ORDER_ID
```
TCP 只实现查询（消息 ID 1001），不包含订阅、事件推送或客户端重连机制。
## 文件与职责
| 文件 | 职责 |
| --- | --- |
| `cmd/hello/main.go` | 第一个独立可运行的最小观察点 |
| `cmd/orders/main.go` | 参数、选择存储、信号、最后关闭数据库 |
| `internal/app/app.go` | 将同一个 Service 接入 HTTP/TCP；启动与等待停机 |
| `internal/httpapi/api.go` | HTTP JSON、路由和错误状态码 |
| `internal/order/service.go` | 模型、业务校验、状态规则、存储接口 |
| `internal/memory/store.go` | map 与互斥锁；用于先理解流程 |
| `internal/sqlite/store.go` | SQL、事务、数据库错误转换 |
| `internal/sqlite/schema.sql` | 初始建表；不是版本化迁移系统 |
| `internal/messageapi/api.go` | 消息 ID 1001 的解码、业务调用、响应 |
| `internal/tcpclient/client.go` | 编码、TCP 读写、响应关联 |
| `cmd/check/main.go` | 对运行中的本地 HTTP 服务执行完整操作 |
| `cmd/ordercli/main.go` | TCP 查询命令行入口 |
| `internal/storetest/contract.go` | 两个存储实现共用的行为测试 |
## 验证
```sh
go test ./... -count=1
go test -race ./... -count=1
go vet ./...
```
覆盖业务校验、数量边界、共享与独立存储、重复取消、两种存储行为、并发取消、
失败回滚、SQLite 重开后数据保持与表约束、HTTP 错误响应、跨 HTTP/TCP 读写、
正常停止与端口占用时的启动失败收尾。测试使用临时数据库，
不会读取或删除你运行示例时的 `orders.db`。
## 有意保留的边界
这是教学应用，不是电商或生产模板。没有用户、鉴权、支付、库存、计价、分页、
创建请求幂等键、消息订阅、监控系统和多实例部署。重复创建会产生两笔订单；
只有重复取消是幂等的。SKU 是字符串，数量范围为 1–20。
SQLite 实现使用单连接串行化本进程的数据库操作，方便理解事务，不能作为吞吐
或多实例承诺。接入 PostgreSQL 时需要真正实现和验证行锁/条件更新、迁移、
连接池、备份恢复，并重新跑存储行为测试。鉴权仍属于应用或网关，KPro 不代做。
建表脚本中的 `CREATE TABLE IF NOT EXISTS` 不会升级已有表结构。
修改 schema 后请使用一个新的**测试数据库路径**，或实现受控的版本迁移；
不要删除已有业务数据库。go.mod
module example.com/orders
go 1.26.9
require (
github.com/saramand9/kpro v0.2.1
modernc.org/sqlite v1.29.10
)
require (
github.com/dustin/go-humanize v1.0.1 // indirect
github.com/google/uuid v1.6.0 // indirect
github.com/hashicorp/golang-lru/v2 v2.0.7 // indirect
github.com/mattn/go-isatty v0.0.20 // indirect
github.com/ncruces/go-strftime v0.1.9 // indirect
github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec // indirect
golang.org/x/sys v0.48.0 // indirect
golang.org/x/time v0.5.0 // indirect
modernc.org/gc/v3 v3.0.0-20240107210532-573471604cb6 // indirect
modernc.org/libc v1.49.3 // indirect
modernc.org/mathutil v1.6.0 // indirect
modernc.org/memory v1.8.0 // indirect
modernc.org/strutil v1.2.0 // indirect
modernc.org/token v1.1.0 // indirect
)go.sum
github.com/andybalholm/brotli v1.2.1 h1:R+f5xP285VArJDRgowrfb9DqL18yVK0gKAW/F+eTWro=
github.com/andybalholm/brotli v1.2.1/go.mod h1:rzTDkvFWvIrjDXZHkuS16NPggd91W3kUSvPlQ1pLaKY=
github.com/dustin/go-humanize v1.0.1 h1:GzkhY7T5VNhEkwH0PVJgjz+fX1rhBrR7pRT3mDkpeCY=
github.com/dustin/go-humanize v1.0.1/go.mod h1:Mu1zIs6XwVuF/gI1OepvI0qD18qycQx+mFykh5fBlto=
github.com/fasthttp/router v1.5.4 h1:oxdThbBwQgsDIYZ3wR1IavsNl6ZS9WdjKukeMikOnC8=
github.com/fasthttp/router v1.5.4/go.mod h1:3/hysWq6cky7dTfzaaEPZGdptwjwx0qzTgFCKEWRjgc=
github.com/google/pprof v0.0.0-20240409012703-83162a5b38cd h1:gbpYu9NMq8jhDVbvlGkMFWCjLFlqqEZjEmObmhUy6Vo=
github.com/google/pprof v0.0.0-20240409012703-83162a5b38cd/go.mod h1:kf6iHlnVGwgKolg33glAes7Yg/8iWP8ukqeldJSO7jw=
github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0=
github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
github.com/hashicorp/golang-lru/v2 v2.0.7 h1:a+bsQ5rvGLjzHuww6tVxozPZFVghXaHOwFs4luLUK2k=
github.com/hashicorp/golang-lru/v2 v2.0.7/go.mod h1:QeFd9opnmA6QUJc5vARoKUSoFhyfM2/ZepoAG6RGpeM=
github.com/klauspost/compress v1.18.7 h1:aUyZsS4kH3QTKurYhAOwAHxllVPnOthb3vPfnF1Ehjw=
github.com/klauspost/compress v1.18.7/go.mod h1:cwPg85FWrGar70rWktvGQj8/hthj3wpl0PGDogxkrSQ=
github.com/klauspost/cpuid/v2 v2.2.9 h1:66ze0taIn2H33fBvCkXuv9BmCwDfafmiIVpKV9kKGuY=
github.com/klauspost/cpuid/v2 v2.2.9/go.mod h1:rqkxqrZ1EhYM9G+hXH7YdowN5R5RGN6NK4QwQ3WMXF8=
github.com/klauspost/reedsolomon v1.10.0 h1:MonMtg979rxSHjwtsla5dZLhreS0Lu42AyQ20bhjIGg=
github.com/klauspost/reedsolomon v1.10.0/go.mod h1:qHMIzMkuZUWqIh8mS/GruPdo3u0qwX2jk/LH440ON7Y=
github.com/mattn/go-isatty v0.0.20 h1:xfD0iDuEKnDkl03q4limB+vH+GxLEtL/jb4xVJSWWEY=
github.com/mattn/go-isatty v0.0.20/go.mod h1:W+V8PltTTMOvKvAeJH7IuucS94S2C6jfK/D7dTCTo3Y=
github.com/ncruces/go-strftime v0.1.9 h1:bY0MQC28UADQmHmaF5dgpLmImcShSi2kHU9XLdhx/f4=
github.com/ncruces/go-strftime v0.1.9/go.mod h1:Fwc5htZGVVkseilnfgOVb9mKy6w1naJmn9CehxcKcls=
github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4=
github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
github.com/quic-go/qpack v0.6.0 h1:g7W+BMYynC1LbYLSqRt8PBg5Tgwxn214ZZR34VIOjz8=
github.com/quic-go/qpack v0.6.0/go.mod h1:lUpLKChi8njB4ty2bFLX2x4gzDqXwUpaO1DP9qMDZII=
github.com/quic-go/quic-go v0.59.1 h1:0Gmua0HW1Tv7ANR7hUYwRyD0MG5OJfgvYSZasGZzBic=
github.com/quic-go/quic-go v0.59.1/go.mod h1:upnsH4Ju1YkqpLXC305eW3yDZ4NfnNbmQRCMWS58IKU=
github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec h1:W09IVJc94icq4NjY3clb7Lk8O1qJ8BdBEF8z0ibU0rE=
github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec/go.mod h1:qqbHyh8v60DhA7CoWK5oRCqLrMHRGoxYCSS9EjAz6Eo=
github.com/saramand9/kpro v0.2.1 h1:AZfqARHOnOeCKaO7kqgHtRNrpwEE8XYyuCUJcQ3ug3s=
github.com/saramand9/kpro v0.2.1/go.mod h1:E7k8aDdclZWh6KCtN/pv95x7D1A/lh4fh46R05w4M6o=
github.com/savsgio/gotils v0.0.0-20240704082632-aef3928b8a38 h1:D0vL7YNisV2yqE55+q0lFuGse6U8lxlg7fYTctlT5Gc=
github.com/savsgio/gotils v0.0.0-20240704082632-aef3928b8a38/go.mod h1:sM7Mt7uEoCeFSCBM+qBrqvEo+/9vdmj19wzp3yzUhmg=
github.com/templexxx/cpufeat v0.0.0-20180724012125-cef66df7f161 h1:89CEmDvlq/F7SJEOqkIdNDGJXrQIhuIx9D2DBXjavSU=
github.com/templexxx/cpufeat v0.0.0-20180724012125-cef66df7f161/go.mod h1:wM7WEvslTq+iOEAMDLSzhVuOt5BRZ05WirO+b09GHQU=
github.com/templexxx/xor v0.0.0-20191217153810-f85b25db303b h1:fj5tQ8acgNUr6O8LEplsxDhUIe2573iLkJc+PqnzZTI=
github.com/templexxx/xor v0.0.0-20191217153810-f85b25db303b/go.mod h1:5XA7W9S6mni3h5uvOC75dA3m9CCCaS83lltmc0ukdi4=
github.com/tjfoc/gmsm v1.4.1 h1:aMe1GlZb+0bLjn+cKTPEvvn9oUEBlJitaZiiBwsbgho=
github.com/tjfoc/gmsm v1.4.1/go.mod h1:j4INPkHWMrhJb38G+J6W4Tw0AbuN8Thu3PbdVYhVcTE=
github.com/valyala/bytebufferpool v1.0.0 h1:GqA5TC/0021Y/b9FG4Oi9Mr3q7XYx6KllzawFIhcdPw=
github.com/valyala/bytebufferpool v1.0.0/go.mod h1:6bBcMArwyJ5K/AmCkWv1jt77kVWyCJ6HpOuEn7z0Csc=
github.com/valyala/fasthttp v1.72.0 h1:R7kYdoWhn1ye1fVpP+cDHDJwYm3NkwLliwgzJ/Abg7M=
github.com/valyala/fasthttp v1.72.0/go.mod h1:zsbLTYqcpIktdQytlVBwIjY9La5d6bs990nBxWg8efk=
github.com/xtaci/kcp-go v5.4.20+incompatible h1:TN1uey3Raw0sTz0Fg8GkfM0uH3YwzhnZWQ1bABv5xAg=
github.com/xtaci/kcp-go v5.4.20+incompatible/go.mod h1:bN6vIwHQbfHaHtFpEssmWsN45a+AZwO7eyRCmEIbtvE=
golang.org/x/crypto v0.57.0 h1:3ZVCjf8Ggz7zneR/EHRVx68Ctf+2pmIMP2UFhh9cC6M=
golang.org/x/crypto v0.57.0/go.mod h1:Fdz0i5U6CoizGwLda9DttjSk6qlZo25zYNtR+ycvuZA=
golang.org/x/mod v0.41.0 h1:qJmnOUb4YB+FsEuM3HcWucdZASCPGhsX6uljO6pog0c=
golang.org/x/mod v0.41.0/go.mod h1:Ek9pY8RKWXwsWvd3rQiHYtMqkjSUV+s1Rj7j4H5Ur6o=
golang.org/x/net v0.60.0 h1:79p50tfZlm0J9YfoDsSi639qSXNGVwEzOPLCxM2FsYU=
golang.org/x/net v0.60.0/go.mod h1:2DA/G1UfVbCpQPeWTmMPGY7Cs2PkBkwu743bVX5PIVg=
golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
golang.org/x/sys v0.48.0 h1:bbX/i/6MgT9BVLM9RT1thmxL04yeTAhbEz4SyadbXoo=
golang.org/x/sys v0.48.0/go.mod h1:hNLxWAXmnKAxqDtdwIYC4bM9oQPEecfsnNMuSxOs3og=
golang.org/x/text v0.42.0 h1:JbOZXgfeCPU9gacVtYliJqOhD+zhrEqK4LfdpmlUZqI=
golang.org/x/text v0.42.0/go.mod h1:ojzP1Z+2QtioaF8DTtO8K5q7JWVVYwZKenzujK0Zd0E=
golang.org/x/time v0.5.0 h1:o7cqy6amK/52YcAKIPlM3a+Fpj35zvRj2TP+e1xFSfk=
golang.org/x/time v0.5.0/go.mod h1:3BpzKBy/shNhVucY/MWOyx10tF3SFh9QdLuxbVysPQM=
golang.org/x/tools v0.19.0 h1:tfGCXNR1OsFG+sVdLAitlpjAvD/I6dHDKnYrpEZUHkw=
golang.org/x/tools v0.19.0/go.mod h1:qoJWxmGSIBmAeriMx19ogtrEPrGtDbPK634QFIcLAhc=
google.golang.org/genproto/googleapis/rpc v0.0.0-20260526163538-3dc84a4a5aaa h1:mZHHdPZl0dbGHCflZgAq/Q468DWVFcU2whhB2KAo8fk=
google.golang.org/genproto/googleapis/rpc v0.0.0-20260526163538-3dc84a4a5aaa/go.mod h1:4Hqkh8ycfw05ld/3BWL7rJOSfebL2Q+DVDeRgYgxUU8=
google.golang.org/grpc v1.83.2 h1:EManeRomTObA0BU7I8vXgg/78uE5MJ9M8B39EX2WscU=
google.golang.org/grpc v1.83.2/go.mod h1:YPI1hK3kDked6iHvgX3tR0y+nX/qpMFKhPgFsokw1S8=
google.golang.org/protobuf v1.36.11 h1:fV6ZwhNocDyBLK0dj+fg8ektcVegBBuEolpbTQyBNVE=
google.golang.org/protobuf v1.36.11/go.mod h1:HTf+CrKn2C3g5S8VImy6tdcUvCska2kB7j23XfzDpco=
modernc.org/cc/v4 v4.20.0 h1:45Or8mQfbUqJOG9WaxvlFYOAQO0lQ5RvqBcFCXngjxk=
modernc.org/cc/v4 v4.20.0/go.mod h1:HM7VJTZbUCR3rV8EYBi9wxnJ0ZBRiGE5OeGXNA0IsLQ=
modernc.org/ccgo/v4 v4.16.0 h1:ofwORa6vx2FMm0916/CkZjpFPSR70VwTjUCe2Eg5BnA=
modernc.org/ccgo/v4 v4.16.0/go.mod h1:dkNyWIjFrVIZ68DTo36vHK+6/ShBn4ysU61So6PIqCI=
modernc.org/fileutil v1.3.0 h1:gQ5SIzK3H9kdfai/5x41oQiKValumqNTDXMvKo62HvE=
modernc.org/fileutil v1.3.0/go.mod h1:XatxS8fZi3pS8/hKG2GH/ArUogfxjpEKs3Ku3aK4JyQ=
modernc.org/gc/v2 v2.4.1 h1:9cNzOqPyMJBvrUipmynX0ZohMhcxPtMccYgGOJdOiBw=
modernc.org/gc/v2 v2.4.1/go.mod h1:wzN5dK1AzVGoH6XOzc3YZ+ey/jPgYHLuVckd62P0GYU=
modernc.org/gc/v3 v3.0.0-20240107210532-573471604cb6 h1:5D53IMaUuA5InSeMu9eJtlQXS2NxAhyWQvkKEgXZhHI=
modernc.org/gc/v3 v3.0.0-20240107210532-573471604cb6/go.mod h1:Qz0X07sNOR1jWYCrJMEnbW/X55x206Q7Vt4mz6/wHp4=
modernc.org/libc v1.49.3 h1:j2MRCRdwJI2ls/sGbeSk0t2bypOG/uvPZUsGQFDulqg=
modernc.org/libc v1.49.3/go.mod h1:yMZuGkn7pXbKfoT/M35gFJOAEdSKdxL0q64sF7KqCDo=
modernc.org/mathutil v1.6.0 h1:fRe9+AmYlaej+64JsEEhoWuAYBkOtQiMEU7n/XgfYi4=
modernc.org/mathutil v1.6.0/go.mod h1:Ui5Q9q1TR2gFm0AQRqQUaBWFLAhQpCwNcuhBOSedWPo=
modernc.org/memory v1.8.0 h1:IqGTL6eFMaDZZhEWwcREgeMXYwmW83LYW8cROZYkg+E=
modernc.org/memory v1.8.0/go.mod h1:XPZ936zp5OMKGWPqbD3JShgd/ZoQ7899TUuQqxY+peU=
modernc.org/opt v0.1.3 h1:3XOZf2yznlhC+ibLltsDGzABUGVx8J6pnFMS3E4dcq4=
modernc.org/opt v0.1.3/go.mod h1:WdSiB5evDcignE70guQKxYUl14mgWtbClRi5wmkkTX0=
modernc.org/sortutil v1.2.0 h1:jQiD3PfS2REGJNzNCMMaLSp/wdMNieTbKX920Cqdgqc=
modernc.org/sortutil v1.2.0/go.mod h1:TKU2s7kJMf1AE84OoiGppNHJwvB753OYfNl2WRb++Ss=
modernc.org/sqlite v1.29.10 h1:3u93dz83myFnMilBGCOLbr+HjklS6+5rJLx4q86RDAg=
modernc.org/sqlite v1.29.10/go.mod h1:ItX2a1OVGgNsFh6Dv60JQvGfJfTPHPVpV6DF59akYOA=
modernc.org/strutil v1.2.0 h1:agBi9dp1I+eOnxXeiZawM8F4LawKv4NzGWSaLfyeNZA=
modernc.org/strutil v1.2.0/go.mod h1:/mdcBmfOibveCTBxUl5B5l6W+TTH1FXPLHZE6bTosX0=
modernc.org/token v1.1.0 h1:Xl7Ap9dKaEs5kLoOQeQmPWevfnk/DM5qcLcYlA8ys6Y=
modernc.org/token v1.1.0/go.mod h1:UGzOrNV1mAFSEB63lOFHIpNRUVMvYTc6yu1SMY/XTDM=cmd/hello/main.go
// This is the first runnable checkpoint: one file, one HTTP route, no database.
package main
import (
"context"
"log"
"net/http"
"os"
"os/signal"
"syscall"
"time"
"github.com/saramand9/kpro/server"
"github.com/saramand9/kpro/transport/httpx"
)
func run() error {
app, err := server.NewChecked(server.Config{Name: "hello", ShutdownTimeout: 5 * time.Second})
if err != nil {
return err
}
endpoint, err := server.MountNativeChecked(app, httpx.New(httpx.Config{Addr: "127.0.0.1:8080"}))
if err != nil {
return err
}
endpoint.Get("/hello", func(c *httpx.Context) error {
return c.JSON(http.StatusOK, map[string]string{"message": "hello KPro"})
})
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
signals := make(chan os.Signal, 1)
signal.Notify(signals, os.Interrupt, syscall.SIGTERM)
defer signal.Stop(signals)
go func() {
select {
case <-signals:
cancel()
case <-ctx.Done():
}
}()
err = app.Run(ctx)
if ctx.Err() != nil && server.IsContextTermination(err) {
return nil
}
return err
}
func main() {
if err := run(); err != nil {
log.Print(err)
os.Exit(1)
}
}cmd/orders/main.go
package main
import (
"context"
"flag"
"log"
"os"
"os/signal"
"syscall"
"time"
"example.com/orders/internal/app"
"example.com/orders/internal/memory"
"example.com/orders/internal/order"
"example.com/orders/internal/sqlite"
)
func main() {
if err := run(); err != nil {
log.Print(err)
os.Exit(1) // run's deferred cleanup has already completed.
}
log.Print("service stopped")
}
func run() error {
httpAddr := flag.String("http", "127.0.0.1:8080", "HTTP listen address")
tcpAddr := flag.String("tcp", "", "optional TCP listen address")
dbPath := flag.String("db", "orders.db", "SQLite file path")
inMemory := flag.Bool("memory", false, "use temporary memory store")
flag.Parse()
var store order.Store
var ready func(context.Context) error
if *inMemory {
m := memory.New()
store, ready = m, m.Ping
} else {
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
db, err := sqlite.Open(ctx, *dbPath)
cancel()
if err != nil {
return err
}
defer db.Close()
store, ready = db, db.Ping
}
runtime, err := app.New(app.Config{HTTPAddr: *httpAddr, TCPAddr: *tcpAddr}, store, ready)
if err != nil {
return err
}
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
signals := make(chan os.Signal, 1)
signal.Notify(signals, os.Interrupt, syscall.SIGTERM)
defer signal.Stop(signals)
// Translate process signals to ordinary cancellation. Go 1.26's
// NotifyContext attaches a signal-specific Cause instead.
go func() {
select {
case <-signals:
cancel()
case <-ctx.Done():
}
}()
return app.Run(ctx, runtime)
}cmd/check/main.go
// check exercises a running LOCAL teaching server. It creates and cancels an
// order; do not point it at a production service.
package main
import (
"bytes"
"encoding/json"
"flag"
"fmt"
"io"
"net/http"
"os"
"time"
"example.com/orders/internal/order"
)
func main() {
base := flag.String("base", "http://127.0.0.1:8080", "local teaching server URL")
flag.Parse()
if err := run(*base); err != nil {
fmt.Fprintln(os.Stderr, err)
os.Exit(1)
}
}
func run(base string) error {
client := &http.Client{Timeout: 3 * time.Second}
defer client.CloseIdleConnections()
call := func(method, path, body string, want int) ([]byte, error) {
req, err := http.NewRequest(method, base+path, bytes.NewBufferString(body))
if err != nil {
return nil, err
}
req.Header.Set("Content-Type", "application/json")
resp, err := client.Do(req)
if err != nil {
return nil, err
}
defer resp.Body.Close()
data, err := io.ReadAll(io.LimitReader(resp.Body, 8192))
if err != nil {
return nil, err
}
fmt.Printf("%s %s -> %d\n%s\n", method, path, resp.StatusCode, data)
if resp.StatusCode != want {
return nil, fmt.Errorf("expected status %d, got %d", want, resp.StatusCode)
}
return data, nil
}
if _, err := call("GET", "/readyz", "", 200); err != nil {
return err
}
body, err := call("POST", "/orders", `{"sku":"coffee","quantity":2}`, 201)
if err != nil {
return err
}
var created order.Order
if err := json.Unmarshal(body, &created); err != nil {
return err
}
for _, step := range []struct {
method, path, body string
status int
}{
{"GET", "/orders/" + created.ID, "", 200},
{"POST", "/orders", `{"sku":"coffee","quantity":0}`, 422},
{"POST", "/orders/" + created.ID + "/cancel", "", 200},
{"POST", "/orders/" + created.ID + "/cancel", "", 200},
} {
if _, err := call(step.method, step.path, step.body, step.status); err != nil {
return err
}
}
fmt.Printf("Order ID for the persistence / TCP exercise: %s\n", created.ID)
return nil
}cmd/ordercli/main.go
package main
import (
"encoding/json"
"flag"
"log"
"os"
"example.com/orders/internal/tcpclient"
)
func main() {
addr := flag.String("addr", "127.0.0.1:8088", "TCP address")
id := flag.String("id", "", "order ID returned by HTTP")
flag.Parse()
if *id == "" {
log.Fatal("pass -id with an existing order ID")
}
o, err := tcpclient.Get(*addr, *id)
if err != nil {
log.Fatal(err)
}
if err := json.NewEncoder(os.Stdout).Encode(o); err != nil {
log.Fatal(err)
}
}internal/app/app.go
// Package app is the composition root: it wires existing parts together.
package app
import (
"context"
"errors"
"log"
"time"
"example.com/orders/internal/httpapi"
"example.com/orders/internal/messageapi"
"example.com/orders/internal/order"
"github.com/saramand9/kpro/codec"
"github.com/saramand9/kpro/core"
"github.com/saramand9/kpro/middleware"
"github.com/saramand9/kpro/server"
"github.com/saramand9/kpro/transport/httpx"
"github.com/saramand9/kpro/transport/tcp"
)
type Config struct {
HTTPAddr string
TCPAddr string // Empty disables TCP.
}
func New(cfg Config, store order.Store, ready func(context.Context) error) (*server.Server, error) {
runtime, err := server.NewChecked(server.Config{
Name: "orders", ShutdownTimeout: 5 * time.Second,
Router: core.RouterConfig{
SyncQueueSize: 16, MaxInflightPerSession: 16,
MaxInflight: 256, AsyncWorkers: 4, AsyncQueueSize: 64,
},
})
if err != nil {
return nil, err
}
runtime.Use(middleware.Recover())
orders := order.New(store) // One instance shared by both entry points.
httpEndpoint, err := server.MountNativeChecked(runtime, httpx.New(httpx.Config{
Addr: cfg.HTTPAddr, MaxBodyBytes: 4 << 10,
ReadHeaderTimeout: 3 * time.Second, ReadTimeout: 5 * time.Second,
WriteTimeout: 5 * time.Second, IdleTimeout: 30 * time.Second,
ShutdownTimeout: 5 * time.Second,
}))
if err != nil {
return nil, err
}
httpapi.Register(httpEndpoint, orders, ready)
if cfg.TCPAddr != "" {
messageEndpoint, err := runtime.MountChecked(tcp.New(tcp.Config{
Addr: cfg.TCPAddr, Codec: codec.KProV1(),
MaxFrameSize: 4 << 10, MaxConnections: 64,
SendQueueSize: 16, MaxPendingBytes: 64 << 10,
ReadTimeout: 30 * time.Second, WriteTimeout: 3 * time.Second,
}))
if err != nil {
return nil, err
}
if err := messageapi.Register(messageEndpoint, orders); err != nil {
return nil, err
}
}
return runtime, nil
}
func Run(parent context.Context, runtime *server.Server) error {
ctx, stop := context.WithCancel(parent)
defer stop()
handle, err := runtime.Start(ctx)
if err != nil {
return err
}
readyCtx, cancelReady := context.WithTimeout(ctx, 5*time.Second)
err = handle.WaitReady(readyCtx)
cancelReady()
if err != nil {
stop()
waitErr := handle.Wait() // Join shutdown before the caller closes the DB.
if server.IsContextTermination(waitErr) {
waitErr = nil
}
return errors.Join(err, waitErr)
}
log.Print("listeners ready")
err = handle.Wait()
if parent.Err() != nil && server.IsContextTermination(err) {
return nil
}
return err
}internal/app/app_test.go
package app_test
import (
"bytes"
"context"
"encoding/json"
"io"
"net"
"net/http"
"strings"
"testing"
"time"
"example.com/orders/internal/app"
"example.com/orders/internal/memory"
"example.com/orders/internal/order"
"example.com/orders/internal/tcpclient"
"github.com/saramand9/kpro/server"
)
func freeAddress(t *testing.T) string {
t.Helper()
listener, err := net.Listen("tcp", "127.0.0.1:0")
if err != nil {
t.Fatal(err)
}
addr := listener.Addr().String()
if err := listener.Close(); err != nil {
t.Fatal(err)
}
return addr
}
func TestHTTPAndTCPShareOrders(t *testing.T) {
httpAddr, tcpAddr := freeAddress(t), freeAddress(t)
store := memory.New()
runtime, err := app.New(app.Config{HTTPAddr: httpAddr, TCPAddr: tcpAddr}, store, store.Ping)
if err != nil {
t.Fatal(err)
}
ctx, stop := context.WithCancel(context.Background())
handle, err := runtime.Start(ctx)
if err != nil {
stop()
t.Fatal(err)
}
t.Cleanup(func() {
stop()
select {
case <-handle.Done():
if err := handle.Wait(); err != nil && !server.IsContextTermination(err) {
t.Errorf("shutdown: %v", err)
}
case <-time.After(7 * time.Second):
t.Error("shutdown did not finish within budget")
}
})
readyCtx, cancel := context.WithTimeout(ctx, 3*time.Second)
defer cancel()
if err := handle.WaitReady(readyCtx); err != nil {
t.Fatal(err)
}
client := &http.Client{Timeout: 3 * time.Second}
defer client.CloseIdleConnections()
resp, err := client.Post("http://"+httpAddr+"/orders", "application/json", bytes.NewBufferString(`{"sku":"coffee","quantity":2}`))
if err != nil {
t.Fatal(err)
}
var created order.Order
err = json.NewDecoder(resp.Body).Decode(&created)
resp.Body.Close()
if err != nil || resp.StatusCode != 201 {
t.Fatalf("create: status %d error %v", resp.StatusCode, err)
}
got, err := tcpclient.Get(tcpAddr, created.ID)
if err != nil || got != created {
t.Fatalf("TCP = %+v, %v; HTTP = %+v", got, err, created)
}
cancelResp, err := client.Post("http://"+httpAddr+"/orders/"+created.ID+"/cancel", "", nil)
if err != nil {
t.Fatal(err)
}
_, _ = io.Copy(io.Discard, cancelResp.Body)
cancelResp.Body.Close()
if cancelResp.StatusCode != 200 {
t.Fatalf("cancel: %d", cancelResp.StatusCode)
}
got, err = tcpclient.Get(tcpAddr, created.ID)
if err != nil || got.Status != order.Cancelled {
t.Fatalf("TCP stale after HTTP cancel: %+v, %v", got, err)
}
if _, err := tcpclient.Get(tcpAddr, "missing"); err == nil || !strings.Contains(err.Error(), "order_not_found") {
t.Fatalf("missing TCP order: %v", err)
}
stop()
select {
case <-handle.Done():
case <-time.After(7 * time.Second):
t.Fatal("server did not stop")
}
if runtime.State() != server.StateStopped {
t.Fatalf("state after shutdown: %v", runtime.State())
}
}
func TestRunRejectsOccupiedPort(t *testing.T) {
occupied, err := net.Listen("tcp", "127.0.0.1:0")
if err != nil {
t.Fatal(err)
}
t.Cleanup(func() { _ = occupied.Close() })
store := memory.New()
runtime, err := app.New(app.Config{HTTPAddr: occupied.Addr().String()}, store, store.Ping)
if err != nil {
t.Fatal(err)
}
ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
defer cancel()
if err := app.Run(ctx, runtime); err == nil || server.IsContextTermination(err) {
t.Fatalf("occupied port should fail startup, got %v", err)
}
if runtime.State() != server.StateStopped {
t.Fatalf("failed startup left runtime in %v", runtime.State())
}
}internal/httpapi/api.go
// Package httpapi translates HTTP requests/responses; business rules live in order.
package httpapi
import (
"bytes"
"context"
"encoding/json"
"errors"
"io"
"log"
"mime"
"net/http"
"time"
"example.com/orders/internal/order"
"github.com/saramand9/kpro/transport/httpx"
)
type API struct{ orders *order.Service }
func Register(endpoint *httpx.Endpoint, orders *order.Service, ready func(context.Context) error) {
api := &API{orders: orders}
endpoint.Get("/livez", func(c *httpx.Context) error {
return c.JSON(http.StatusOK, map[string]string{"status": "alive"})
})
endpoint.Get("/readyz", func(c *httpx.Context) error {
ctx, cancel := context.WithTimeout(c.Context(), time.Second)
defer cancel()
if err := ready(ctx); err != nil {
return c.JSON(http.StatusServiceUnavailable, map[string]string{"error": "not_ready"})
}
return c.JSON(http.StatusOK, map[string]string{"status": "ready"})
})
endpoint.Post("/orders", api.create)
endpoint.Get("/orders/{id}", api.get)
endpoint.Post("/orders/{id}/cancel", api.cancel)
}
func (a *API) create(c *httpx.Context) error {
mediaType, _, err := mime.ParseMediaType(c.Request().Header.Get("Content-Type"))
if err != nil || mediaType != "application/json" {
return c.JSON(http.StatusUnsupportedMediaType, map[string]string{"error": "use_application_json"})
}
body, err := c.Body()
if err != nil {
var tooLarge *http.MaxBytesError
if errors.As(err, &tooLarge) {
return c.JSON(http.StatusRequestEntityTooLarge, map[string]string{"error": "body_too_large"})
}
return c.JSON(http.StatusBadRequest, map[string]string{"error": "invalid_body"})
}
var input order.CreateInput
decoder := json.NewDecoder(bytes.NewReader(body))
decoder.DisallowUnknownFields()
if err := decoder.Decode(&input); err != nil {
return c.JSON(http.StatusBadRequest, map[string]string{"error": "invalid_json"})
}
if err := decoder.Decode(new(any)); err != io.EOF {
return c.JSON(http.StatusBadRequest, map[string]string{"error": "one_json_object_required"})
}
ctx, cancel := context.WithTimeout(c.Context(), 2*time.Second)
defer cancel()
o, err := a.orders.Create(ctx, input)
if err != nil {
return respondError(c, err)
}
c.Header().Set("Location", "/orders/"+o.ID)
return c.JSON(http.StatusCreated, o)
}
func (a *API) get(c *httpx.Context) error {
ctx, cancel := context.WithTimeout(c.Context(), 2*time.Second)
defer cancel()
o, err := a.orders.Get(ctx, c.Param("id"))
if err != nil {
return respondError(c, err)
}
return c.JSON(http.StatusOK, o)
}
func (a *API) cancel(c *httpx.Context) error {
ctx, cancel := context.WithTimeout(c.Context(), 2*time.Second)
defer cancel()
o, err := a.orders.Cancel(ctx, c.Param("id"))
if err != nil {
return respondError(c, err)
}
return c.JSON(http.StatusOK, o)
}
func respondError(c *httpx.Context, err error) error {
status, code := http.StatusInternalServerError, "internal_error"
switch {
case errors.Is(err, order.ErrInvalid):
status, code = http.StatusUnprocessableEntity, "invalid_order"
case errors.Is(err, order.ErrNotFound):
status, code = http.StatusNotFound, "order_not_found"
case errors.Is(err, order.ErrConflict):
status, code = http.StatusConflict, "order_conflict"
case errors.Is(err, context.DeadlineExceeded), errors.Is(err, context.Canceled):
status, code = http.StatusServiceUnavailable, "request_unavailable"
default:
// Details go to server logs, never into the public response.
log.Printf("order operation failed: %v", err)
}
return c.JSON(status, map[string]string{"error": code})
}internal/httpapi/api_test.go
package httpapi_test
import (
"context"
"encoding/json"
"errors"
"net/http"
"net/http/httptest"
"strings"
"testing"
"example.com/orders/internal/httpapi"
"example.com/orders/internal/memory"
"example.com/orders/internal/order"
"github.com/saramand9/kpro/transport/httpx"
)
func endpoint() *httpx.Endpoint {
db := memory.New()
e := httpx.New(httpx.Config{MaxBodyBytes: 4096})
httpapi.Register(e, order.New(db), db.Ping)
return e
}
func request(e http.Handler, method, path, body string) *httptest.ResponseRecorder {
r := httptest.NewRequest(method, path, strings.NewReader(body))
r.Header.Set("Content-Type", "application/json")
w := httptest.NewRecorder()
e.ServeHTTP(w, r)
return w
}
func TestHTTPJourney(t *testing.T) {
e := endpoint()
created := request(e, "POST", "/orders", `{"sku":"coffee","quantity":2}`)
if created.Code != 201 {
t.Fatalf("create: %d %s", created.Code, created.Body)
}
var o order.Order
if err := json.Unmarshal(created.Body.Bytes(), &o); err != nil {
t.Fatal(err)
}
if created.Header().Get("Location") != "/orders/"+o.ID {
t.Fatal("missing Location header")
}
if got := request(e, "GET", "/orders/"+o.ID, ""); got.Code != 200 {
t.Fatalf("get: %d %s", got.Code, got.Body)
}
first := request(e, "POST", "/orders/"+o.ID+"/cancel", "")
second := request(e, "POST", "/orders/"+o.ID+"/cancel", "")
if first.Code != 200 || second.Code != 200 || first.Body.String() != second.Body.String() || !strings.Contains(second.Body.String(), `"cancelled"`) {
t.Fatalf("cancel: %s / %s", first.Body, second.Body)
}
}
func TestHTTPQuantityBoundary(t *testing.T) {
for _, body := range []string{
`{"sku":"coffee","quantity":1}`,
`{"sku":"coffee","quantity":20}`,
} {
got := request(endpoint(), "POST", "/orders", body)
if got.Code != 201 {
t.Fatalf("create %s: got %d %s; want 201", body, got.Code, got.Body)
}
}
}
func TestHTTPFailures(t *testing.T) {
cases := []struct {
name, body string
status int
}{
{"broken json", `{`, 400},
{"unknown field", `{"sku":"tea","quantity":1,"price":0}`, 400},
{"wrong type", `{"sku":"tea","quantity":"1"}`, 400},
{"trailing json", `{"sku":"tea","quantity":1}{}`, 400},
{"empty sku", `{"sku":"","quantity":1}`, 422},
{"zero quantity", `{"sku":"tea","quantity":0}`, 422},
{"large quantity", `{"sku":"tea","quantity":21}`, 422},
{"large body", strings.Repeat("x", 4097), 413},
}
for _, tc := range cases {
t.Run(tc.name, func(t *testing.T) {
got := request(endpoint(), "POST", "/orders", tc.body)
if got.Code != tc.status {
t.Fatalf("got %d %s; want %d", got.Code, got.Body, tc.status)
}
})
}
if w := request(endpoint(), "GET", "/orders/missing", ""); w.Code != 404 {
t.Fatalf("missing order: %d %s", w.Code, w.Body)
}
r := httptest.NewRequest("POST", "/orders", strings.NewReader("{}"))
w := httptest.NewRecorder()
endpoint().ServeHTTP(w, r)
if w.Code != 415 {
t.Fatalf("content type: %d", w.Code)
}
}
type failedStore struct{ order.Store }
func (failedStore) Create(context.Context, order.Order) error {
return errors.New("private database details")
}
func TestInternalErrorIsNotExposed(t *testing.T) {
e := httpx.New(httpx.Config{})
httpapi.Register(e, order.New(failedStore{}), func(context.Context) error { return errors.New("offline") })
w := request(e, "POST", "/orders", `{"sku":"tea","quantity":1}`)
if w.Code != 500 || strings.Contains(w.Body.String(), "private") {
t.Fatalf("unsafe response: %d %s", w.Code, w.Body)
}
if w := request(e, "GET", "/readyz", ""); w.Code != 503 {
t.Fatalf("readiness = %d", w.Code)
}
if w := request(e, "GET", "/livez", ""); w.Code != 200 {
t.Fatalf("liveness = %d", w.Code)
}
}internal/order/service.go
// Package order contains application rules. It has no KPro or SQL dependency.
package order
import (
"context"
"crypto/rand"
"encoding/hex"
"errors"
"strings"
"time"
)
var (
ErrInvalid = errors.New("sku must be 1..64 characters and quantity must be 1..20")
ErrNotFound = errors.New("order not found")
ErrConflict = errors.New("order cannot be cancelled in its current state")
)
const (
Pending = "pending"
Cancelled = "cancelled"
)
type Order struct {
ID string `json:"id"`
SKU string `json:"sku"`
Quantity int `json:"quantity"`
Status string `json:"status"`
CreatedAt time.Time `json:"created_at"`
}
type CreateInput struct {
SKU string `json:"sku"`
Quantity int `json:"quantity"`
}
// Store describes what the application needs, not how a database works.
type Store interface {
Create(context.Context, Order) error
Get(context.Context, string) (Order, error)
// UpdateStatus must read, apply change, and save atomically. If change fails,
// nothing is saved. change must not do network I/O or call Store again.
UpdateStatus(context.Context, string, func(string) (string, error)) (Order, error)
}
type Service struct{ store Store }
func New(store Store) *Service { return &Service{store: store} }
func (s *Service) Create(ctx context.Context, input CreateInput) (Order, error) {
input.SKU = strings.TrimSpace(input.SKU)
if n := len([]rune(input.SKU)); n == 0 || n > 64 || input.Quantity < 1 || input.Quantity > 20 {
return Order{}, ErrInvalid
}
var id [16]byte
if _, err := rand.Read(id[:]); err != nil {
return Order{}, err
}
o := Order{
ID: hex.EncodeToString(id[:]), SKU: input.SKU, Quantity: input.Quantity,
Status: Pending, CreatedAt: time.Now().UTC(),
}
if err := s.store.Create(ctx, o); err != nil {
return Order{}, err
}
return o, nil
}
func (s *Service) Get(ctx context.Context, id string) (Order, error) {
return s.store.Get(ctx, id)
}
func (s *Service) Cancel(ctx context.Context, id string) (Order, error) {
return s.store.UpdateStatus(ctx, id, func(status string) (string, error) {
switch status {
case Pending:
return Cancelled, nil
case Cancelled:
// A retry returns the same state: cancelling twice is harmless.
return Cancelled, nil
default:
return "", ErrConflict
}
})
}internal/order/service_test.go
package order_test
import (
"context"
"errors"
"strings"
"testing"
"example.com/orders/internal/memory"
"example.com/orders/internal/order"
)
func TestCreate(t *testing.T) {
service := order.New(memory.New())
o, err := service.Create(context.Background(), order.CreateInput{SKU: " coffee ", Quantity: 2})
if err != nil {
t.Fatal(err)
}
if o.ID == "" || o.SKU != "coffee" || o.Quantity != 2 || o.Status != order.Pending || o.CreatedAt.IsZero() {
t.Fatalf("unexpected order: %+v", o)
}
got, err := service.Get(context.Background(), o.ID)
if err != nil || got != o {
t.Fatalf("saved order = %+v, %v; want %+v", got, err, o)
}
}
func TestRejectInvalidOrder(t *testing.T) {
inputs := []order.CreateInput{
{SKU: "", Quantity: 1},
{SKU: " ", Quantity: 1},
{SKU: strings.Repeat("字", 65), Quantity: 1},
{SKU: "coffee", Quantity: -1},
{SKU: "coffee", Quantity: 0},
{SKU: "coffee", Quantity: 21},
}
for _, input := range inputs {
if _, err := order.New(memory.New()).Create(context.Background(), input); !errors.Is(err, order.ErrInvalid) {
t.Errorf("Create(%+v) = %v; want ErrInvalid", input, err)
}
}
}
func TestQuantityBoundary(t *testing.T) {
service := order.New(memory.New())
for _, qty := range []int{1, 20} {
if _, err := service.Create(context.Background(), order.CreateInput{SKU: "coffee", Quantity: qty}); err != nil {
t.Errorf("quantity %d should be accepted: %v", qty, err)
}
}
}
func TestCancelTwice(t *testing.T) {
ctx := context.Background()
service := order.New(memory.New())
o, err := service.Create(ctx, order.CreateInput{SKU: "tea", Quantity: 1})
if err != nil {
t.Fatal(err)
}
first, err := service.Cancel(ctx, o.ID)
if err != nil {
t.Fatal(err)
}
second, err := service.Cancel(ctx, o.ID)
if err != nil || first != second || second.Status != order.Cancelled {
t.Fatalf("retry changed result: first=%+v second=%+v error=%v", first, second, err)
}
}
func TestMissingOrder(t *testing.T) {
service := order.New(memory.New())
if _, err := service.Get(context.Background(), "missing"); !errors.Is(err, order.ErrNotFound) {
t.Fatalf("Get error = %v", err)
}
if _, err := service.Cancel(context.Background(), "missing"); !errors.Is(err, order.ErrNotFound) {
t.Fatalf("Cancel error = %v", err)
}
}
// Separate services share data only when they receive the same store object.
func TestServicesShareStore(t *testing.T) {
ctx := context.Background()
store := memory.New()
writer := order.New(store)
reader := order.New(store)
isolated := order.New(memory.New())
created, err := writer.Create(ctx, order.CreateInput{SKU: "coffee", Quantity: 2})
if err != nil {
t.Fatal(err)
}
got, err := reader.Get(ctx, created.ID)
if err != nil || got != created {
t.Fatalf("shared store: got %+v, %v; want %+v", got, err, created)
}
if _, err := isolated.Get(ctx, created.ID); !errors.Is(err, order.ErrNotFound) {
t.Fatalf("separate store: got %v; want ErrNotFound", err)
}
}internal/memory/store.go
package memory
import (
"context"
"errors"
"sync"
"example.com/orders/internal/order"
)
type Store struct {
mu sync.Mutex
orders map[string]order.Order
}
var _ order.Store = (*Store)(nil)
func New() *Store { return &Store{orders: make(map[string]order.Order)} }
func (s *Store) Create(ctx context.Context, o order.Order) error {
s.mu.Lock()
defer s.mu.Unlock()
if err := ctx.Err(); err != nil {
return err
}
if _, exists := s.orders[o.ID]; exists {
return errors.New("duplicate order id")
}
s.orders[o.ID] = o
return nil
}
func (s *Store) Get(ctx context.Context, id string) (order.Order, error) {
s.mu.Lock()
defer s.mu.Unlock()
if err := ctx.Err(); err != nil {
return order.Order{}, err
}
o, ok := s.orders[id]
if !ok {
return order.Order{}, order.ErrNotFound
}
return o, nil
}
func (s *Store) UpdateStatus(ctx context.Context, id string, change func(string) (string, error)) (order.Order, error) {
s.mu.Lock()
defer s.mu.Unlock()
if err := ctx.Err(); err != nil {
return order.Order{}, err
}
o, ok := s.orders[id]
if !ok {
return order.Order{}, order.ErrNotFound
}
status, err := change(o.Status)
if err != nil {
return order.Order{}, err
}
o.Status = status
s.orders[id] = o
return o, nil
}
func (s *Store) Ping(ctx context.Context) error { return ctx.Err() }internal/memory/store_test.go
package memory_test
import (
"testing"
"example.com/orders/internal/memory"
"example.com/orders/internal/storetest"
)
func TestStoreContract(t *testing.T) { storetest.Run(t, memory.New()) }internal/sqlite/store.go
// Package sqlite implements the application's Store using a local SQLite file.
package sqlite
import (
"context"
"database/sql"
_ "embed"
"errors"
"time"
"example.com/orders/internal/order"
_ "modernc.org/sqlite"
)
//go:embed schema.sql
var schema string
type Store struct{ db *sql.DB }
var _ order.Store = (*Store)(nil)
func Open(ctx context.Context, path string) (*Store, error) {
db, err := sql.Open("sqlite", path)
if err != nil {
return nil, err
}
// This teaching service uses one connection and one process. Operations
// queue here; this is NOT a multi-instance/high-throughput database design.
db.SetMaxOpenConns(1)
if _, err = db.ExecContext(ctx, schema); err != nil {
_ = db.Close()
return nil, err
}
return &Store{db: db}, nil
}
func (s *Store) Close() error { return s.db.Close() }
func (s *Store) Ping(ctx context.Context) error { return s.db.PingContext(ctx) }
func (s *Store) Create(ctx context.Context, o order.Order) error {
_, err := s.db.ExecContext(ctx,
`INSERT INTO orders (id, sku, quantity, status, created_at) VALUES (?, ?, ?, ?, ?)`,
o.ID, o.SKU, o.Quantity, o.Status, o.CreatedAt.Format(time.RFC3339Nano))
return err
}
func (s *Store) Get(ctx context.Context, id string) (order.Order, error) {
return scan(s.db.QueryRowContext(ctx,
`SELECT id, sku, quantity, status, created_at FROM orders WHERE id = ?`, id))
}
func (s *Store) UpdateStatus(ctx context.Context, id string, change func(string) (string, error)) (order.Order, error) {
tx, err := s.db.BeginTx(ctx, nil)
if err != nil {
return order.Order{}, err
}
defer tx.Rollback()
current, err := scan(tx.QueryRowContext(ctx,
`SELECT id, sku, quantity, status, created_at FROM orders WHERE id = ?`, id))
if err != nil {
return order.Order{}, err
}
status, err := change(current.Status)
if err != nil {
return order.Order{}, err
}
// The application currently changes only status. Other fields are immutable.
if _, err = tx.ExecContext(ctx, `UPDATE orders SET status = ? WHERE id = ?`, status, id); err != nil {
return order.Order{}, err
}
if err := tx.Commit(); err != nil {
return order.Order{}, err
}
current.Status = status
return current, nil
}
func scan(row *sql.Row) (order.Order, error) {
var o order.Order
var created string
if err := row.Scan(&o.ID, &o.SKU, &o.Quantity, &o.Status, &created); err != nil {
if errors.Is(err, sql.ErrNoRows) {
return order.Order{}, order.ErrNotFound
}
return order.Order{}, err
}
t, err := time.Parse(time.RFC3339Nano, created)
o.CreatedAt = t
return o, err
}internal/sqlite/store_test.go
package sqlite_test
import (
"context"
"fmt"
"path/filepath"
"testing"
"time"
"example.com/orders/internal/order"
"example.com/orders/internal/sqlite"
"example.com/orders/internal/storetest"
)
func TestStoreContract(t *testing.T) {
db, err := sqlite.Open(context.Background(), filepath.Join(t.TempDir(), "orders.db"))
if err != nil {
t.Fatal(err)
}
t.Cleanup(func() { _ = db.Close() })
storetest.Run(t, db)
}
func TestPersistsAfterReopen(t *testing.T) {
ctx := context.Background()
path := filepath.Join(t.TempDir(), "orders.db")
db, err := sqlite.Open(ctx, path)
if err != nil {
t.Fatal(err)
}
service := order.New(db)
created, err := service.Create(ctx, order.CreateInput{SKU: "coffee", Quantity: 2})
if err != nil {
_ = db.Close()
t.Fatal(err)
}
cancelled, err := service.Cancel(ctx, created.ID)
if err != nil {
_ = db.Close()
t.Fatal(err)
}
if err := db.Close(); err != nil {
t.Fatal(err)
}
reopened, err := sqlite.Open(ctx, path)
if err != nil {
t.Fatal(err)
}
defer reopened.Close()
got, err := order.New(reopened).Get(ctx, created.ID)
if err != nil || got != cancelled {
t.Fatalf("reopened order = %+v, %v; want %+v", got, err, cancelled)
}
}
// Bypass Service deliberately: the schema also rejects an invalid new row.
func TestQuantityConstraint(t *testing.T) {
ctx := context.Background()
db, err := sqlite.Open(ctx, filepath.Join(t.TempDir(), "bounds.db"))
if err != nil {
t.Fatal(err)
}
t.Cleanup(func() { _ = db.Close() })
for _, tc := range []struct {
quantity int
accepted bool
}{{0, false}, {1, true}, {20, true}, {21, false}} {
o := order.Order{
ID: fmt.Sprintf("quantity-%d", tc.quantity), SKU: "coffee", Quantity: tc.quantity,
Status: order.Pending, CreatedAt: time.Now().UTC(),
}
err := db.Create(ctx, o)
if (err == nil) != tc.accepted {
t.Errorf("quantity %d: error %v; accepted=%v", tc.quantity, err, tc.accepted)
}
}
}internal/sqlite/schema.sql
CREATE TABLE IF NOT EXISTS orders (
id TEXT PRIMARY KEY,
sku TEXT NOT NULL,
quantity INTEGER NOT NULL CHECK (quantity BETWEEN 1 AND 20),
status TEXT NOT NULL CHECK (status IN ('pending', 'cancelled')),
created_at TEXT NOT NULL
);internal/messageapi/api.go
// Package messageapi adapts one binary message to the SAME order.Service.
package messageapi
import (
"context"
"encoding/json"
"errors"
"log"
"time"
"example.com/orders/internal/order"
"github.com/saramand9/kpro/core"
"github.com/saramand9/kpro/server"
)
const GetOrderID uint32 = 1001
type GetRequest struct {
ID string `json:"id"`
}
type Response struct {
Order *order.Order `json:"order,omitempty"`
Error string `json:"error,omitempty"`
}
func Register(endpoint *server.Endpoint, orders *order.Service) error {
return endpoint.HandleChecked(GetOrderID, func(c core.Context) error {
var req GetRequest
if err := json.Unmarshal(c.Message().Data, &req); err != nil || req.ID == "" {
return reply(c, Response{Error: "invalid_request"})
}
ctx, cancel := context.WithTimeout(core.ContextOf(c), 2*time.Second)
defer cancel()
o, err := orders.Get(ctx, req.ID)
if errors.Is(err, order.ErrNotFound) {
return reply(c, Response{Error: "order_not_found"})
}
if err != nil {
log.Printf("message lookup failed: %v", err)
return reply(c, Response{Error: "request_unavailable"})
}
return reply(c, Response{Order: &o})
})
}
func reply(c core.Context, response Response) error {
data, err := json.Marshal(response)
if err != nil {
return err
}
// Reply preserves the request sequence number. A successful return means
// accepted for sending, not that the remote application has processed it.
return c.Reply(GetOrderID, data)
}internal/tcpclient/client.go
package tcpclient
import (
"bytes"
"encoding/json"
"fmt"
"io"
"net"
"time"
"example.com/orders/internal/messageapi"
"example.com/orders/internal/order"
"github.com/saramand9/kpro/codec"
"github.com/saramand9/kpro/core"
)
func Get(addr, id string) (order.Order, error) {
conn, err := net.DialTimeout("tcp", addr, 3*time.Second)
if err != nil {
return order.Order{}, err
}
defer conn.Close()
if err := conn.SetDeadline(time.Now().Add(3 * time.Second)); err != nil {
return order.Order{}, err
}
body, err := json.Marshal(messageapi.GetRequest{ID: id})
if err != nil {
return order.Order{}, err
}
frame, err := codec.KProV1().Encode(core.Message{ID: messageapi.GetOrderID, Seq: 1, Data: body})
if err != nil {
return order.Order{}, err
}
if _, err := io.Copy(conn, bytes.NewReader(frame)); err != nil {
return order.Order{}, err
}
result, err := codec.KProV1().Read(conn, 4<<10)
if err != nil {
return order.Order{}, err
}
if result.Message.ID != messageapi.GetOrderID || result.Message.Seq != 1 {
return order.Order{}, fmt.Errorf("unexpected reply id/sequence")
}
var response messageapi.Response
if err := json.Unmarshal(result.Message.Data, &response); err != nil {
return order.Order{}, err
}
if response.Error != "" {
return order.Order{}, fmt.Errorf("server: %s", response.Error)
}
if response.Order == nil {
return order.Order{}, fmt.Errorf("missing order in response")
}
return *response.Order, nil
}internal/storetest/contract.go
// Package storetest checks the same application contract against both stores.
package storetest
import (
"context"
"errors"
"sync"
"testing"
"example.com/orders/internal/order"
)
func Run(t *testing.T, store order.Store) {
t.Helper()
ctx := context.Background()
service := order.New(store)
o, err := service.Create(ctx, order.CreateInput{SKU: "tea", Quantity: 1})
if err != nil {
t.Fatal(err)
}
t.Run("duplicate", func(t *testing.T) {
if err := store.Create(ctx, o); err == nil {
t.Fatal("duplicate ID was accepted")
}
})
t.Run("rollback", func(t *testing.T) {
rejected := errors.New("rejected change")
_, err := store.UpdateStatus(ctx, o.ID, func(string) (string, error) {
return order.Cancelled, rejected
})
if !errors.Is(err, rejected) {
t.Fatalf("error = %v", err)
}
got, err := store.Get(ctx, o.ID)
if err != nil || got != o {
t.Fatalf("failed update modified order: %+v, %v", got, err)
}
})
t.Run("cancelled_context", func(t *testing.T) {
cancelled, cancel := context.WithCancel(ctx)
cancel()
if _, err := service.Get(cancelled, o.ID); !errors.Is(err, context.Canceled) {
t.Fatalf("error = %v", err)
}
})
t.Run("concurrent_cancel", func(t *testing.T) {
var wg sync.WaitGroup
for range 20 {
wg.Go(func() {
got, err := service.Cancel(ctx, o.ID)
if err != nil || got.Status != order.Cancelled {
t.Errorf("cancel = %+v, %v", got, err)
}
})
}
wg.Wait()
})
t.Run("not_found", func(t *testing.T) {
if _, err := service.Get(ctx, "missing"); !errors.Is(err, order.ErrNotFound) {
t.Fatalf("error = %v", err)
}
})
}requests/create.json
{"sku":"coffee","quantity":2}requests/invalid.json
{"sku":"coffee","quantity":0}