convertinterface 보완

Reviewed-on: #4

go.mod

@@ -1,6 +1,8 @@
 module repositories.action2quare.com/ayo/gocommon
 
-go 1.20
+go 1.22
+
+toolchain go1.22.4
 
 require (
 	github.com/awa/go-iap v1.32.0
@@ -34,6 +36,7 @@ require (
 	github.com/klauspost/compress v1.13.6 // indirect
 	github.com/matttproud/golang_protobuf_extensions v1.0.4 // indirect
 	github.com/montanaflynn/stats v0.0.0-20171201202039-1bf9dbcd8cbe // indirect
+	github.com/opensearch-project/opensearch-go/v4 v4.5.0 // indirect
 	github.com/pkg/errors v0.9.1 // indirect
 	github.com/prometheus/client_model v0.4.1-0.20230718164431-9a2bf3000d16 // indirect
 	github.com/prometheus/common v0.44.0 // indirect

go.sum

@@ -90,6 +90,8 @@ github.com/montanaflynn/stats v0.0.0-20171201202039-1bf9dbcd8cbe/go.mod h1:wL8QJ
 github.com/nxadm/tail v1.4.8 h1:nPr65rt6Y5JFSKQO7qToXr7pePgD6Gwiw05lkbyAQTE=
 github.com/onsi/ginkgo v1.16.5 h1:8xi0RTUf59SOSfEtZMvwTvXYMzG4gV23XVHOZiXNtnE=
 github.com/onsi/gomega v1.18.1 h1:M1GfJqGRrBrrGGsbxzV5dqM2U2ApXefZCQpkukxYRLE=
+github.com/opensearch-project/opensearch-go/v4 v4.5.0 h1:26XckmmF6MhlXt91Bu1yY6R51jy1Ns/C3XgIfvyeTRo=
+github.com/opensearch-project/opensearch-go/v4 v4.5.0/go.mod h1:VmFc7dqOEM3ZtLhrpleOzeq+cqUgNabqQG5gX0xId64=
 github.com/pires/go-proxyproto v0.7.0 h1:IukmRewDQFWC7kfnb66CSomk2q/seBuilHBYFwyq0Hs=
 github.com/pires/go-proxyproto v0.7.0/go.mod h1:Vz/1JPY/OACxWGQNIRY2BeyDmpoaWmEP40O9LbuiFR4=
 github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4=

logger/logger.go

@@ -2,43 +2,40 @@ package logger
 
 import (
 	"fmt"
-	"io"
 	"log"
 	"os"
-	"path"
 	"runtime"
 	"runtime/debug"
+	"strconv"
 	"strings"
 
 	"repositories.action2quare.com/ayo/gocommon/flagx"
 )
 
 var stdlogger *log.Logger
-var UseLogFile = flagx.Bool("logfile", false, "")
+var _ = flagx.Int("logprefix", 3, "0 : no_prefix, 1 : date, 2 : time, 3 : datetime")
 
 func init() {
 	binpath, _ := os.Executable()
-	binname := path.Base(strings.ReplaceAll(binpath, "\\", "/"))
+	outWriter := os.Stdout
 
-	var outWriter io.Writer
-	outWriter = os.Stdout
-
-	if *UseLogFile {
-		ext := path.Ext(binname)
-		if len(ext) > 0 {
-			binname = binname[:len(binname)-len(ext)]
+	args := os.Args
+	logprefix := 3
+	for _, arg := range args {
+		if strings.HasPrefix(arg, "-logprefix=") {
+			logprefix, _ = strconv.Atoi(arg[11:])
+			break
 		}
-
-		logFile, err := os.Create(fmt.Sprintf("%s.log", binname))
-		if err != nil {
-			os.Stdout.Write([]byte(err.Error()))
-			panic(err)
-		}
-
-		outWriter = io.MultiWriter(outWriter, logFile)
 	}
 
-	stdlogger = log.New(outWriter, "", log.LstdFlags)
+	pid := fmt.Sprintf("[%d]", os.Getpid())
+	outWriter.Write([]byte(strings.Join(append([]string{pid, binpath}, args...), " ")))
+
+	if logprefix < 4 {
+		stdlogger = log.New(outWriter, "", logprefix)
+	} else {
+		stdlogger = log.New(outWriter, "", log.LstdFlags)
+	}
 }
 
 func Println(v ...interface{}) {
@@ -130,3 +127,33 @@ func ErrorWithCallStack(err error) error {
 		frames: frames,
 	}
 }
+
+func ErrorSmallStack() {
+	buf := make([]byte, 1024)
+	n := runtime.Stack(buf, false)
+	if n < len(buf) {
+		buf = buf[:n]
+	}
+
+	Error(string(buf))
+}
+
+func RecoverAndErrorSmallStack(r any) any {
+	if r != nil {
+		pc := make([]uintptr, 10)
+		runtime.Callers(1, pc)
+		curframes := runtime.CallersFrames(pc)
+		var out []string
+		for {
+			frame, more := curframes.Next()
+			out = append(out, fmt.Sprintf("%s\n\t%s:%d", frame.Function, frame.File, frame.Line))
+			if !more {
+				break
+			}
+		}
+
+		Error(strings.Join(out, "\n"))
+	}
+
+	return r
+}

metric/common.go

@@ -2,13 +2,12 @@ package metric
 
 import (
 	"crypto/md5"
-	"encoding/binary"
 	"encoding/hex"
 	"encoding/json"
 	"fmt"
-	"math"
 	"os"
 	"path"
+	"runtime"
 	"sort"
 	"strings"
 	"sync/atomic"
@@ -16,8 +15,6 @@ import (
 	"repositories.action2quare.com/ayo/gocommon/logger"
 )
 
-const metric_value_line_size = 19
-
 type MetricDescription struct {
 	Key         string
 	Type        MetricType
@@ -29,7 +26,35 @@ type MetricDescription struct {
 type Exporter interface {
 	RegisterMetric(*MetricDescription)
 	UpdateMetric(string, float64)
-	Shutdown()
+}
+
+type MetricPipe struct {
+	pipe *os.File
+}
+
+func (mp MetricPipe) Close() {
+	if mp.pipe != nil {
+		mp.pipe.Close()
+		mp.pipe = nil
+	}
+}
+
+func (mp MetricPipe) writeLine(line string) {
+	mp.pipe.WriteString(line + "\n")
+}
+
+func NewMetricPipe(pipeName string) MetricPipe {
+	switch runtime.GOOS {
+	case "linux":
+		pipeName = "/tmp/" + pipeName
+	case "windows":
+		pipeName = `\\.\pipe\` + pipeName
+	}
+
+	f, _ := os.Open(pipeName)
+	return MetricPipe{
+		pipe: f,
+	}
 }
 
 type MetricWriter interface {
@@ -45,13 +70,14 @@ func (mw *metric_empty) Add(int64) {}
 var MetricWriterNil = MetricWriter(&metric_empty{})
 
 type metric_int64 struct {
+	key    string
 	valptr *int64
-	buff   [metric_value_line_size]byte
+	pipe   MetricPipe
 }
 
 func (mw *metric_int64) printOut() {
-	binary.LittleEndian.PutUint64(mw.buff[9:], math.Float64bits(float64(atomic.LoadInt64(mw.valptr))))
-	os.Stdout.Write(mw.buff[:])
+	loaded := atomic.LoadInt64(mw.valptr)
+	mw.pipe.writeLine(fmt.Sprintf("%s:%d", mw.key, loaded))
 }
 
 func (mw *metric_int64) Set(newval int64) {
@@ -64,11 +90,16 @@ func (mw *metric_int64) Add(inc int64) {
 	mw.printOut()
 }
 
-func NewMetric(mt MetricType, name string, help string, constLabels map[string]string) (writer MetricWriter) {
+func NewMetric(pipe MetricPipe, mt MetricType, name string, help string, constLabels map[string]string) (writer MetricWriter) {
 	if !metricEnabled {
 		return MetricWriterNil
 	}
 
+	if constLabels == nil {
+		constLabels = map[string]string{}
+	}
+
+	constLabels["pid"] = fmt.Sprintf("%d", os.Getpid())
 	var disorder []struct {
 		k string
 		v string
@@ -101,35 +132,17 @@ func NewMetric(mt MetricType, name string, help string, constLabels map[string]s
 	})
 
 	impl := &metric_int64{
+		key:    key,
 		valptr: new(int64),
+		pipe:   pipe,
 	}
 
-	impl.buff[0] = METRIC_HEAD_INLINE
-	impl.buff[17] = METRIC_TAIL_INLINE
-	impl.buff[18] = '\n'
-	copy(impl.buff[1:], []byte(key))
-
-	output := append([]byte{METRIC_HEAD_INLINE}, temp...)
-	output = append(output, METRIC_TAIL_INLINE, '\n')
-	os.Stdout.Write(output)
-
+	pipe.writeLine(string(temp))
 	// writer
 
 	return impl
 }
 
-func ReadMetricValue(line []byte) (string, float64) {
-	if len(line) < 16 {
-		return "", 0
-	}
-
-	key := string(line[0:8])
-	valbits := binary.LittleEndian.Uint64(line[8:])
-	val := math.Float64frombits(valbits)
-
-	return key, val
-}
-
 var metricEnabled = false
 
 func init() {

metric/prometheus.go

@@ -1,7 +1,7 @@
 package metric
 
 import (
-	"context"
+	"maps"
 	"math"
 	"sync/atomic"
 
@@ -34,11 +34,6 @@ func convertValueType(in MetricType) prometheus.ValueType {
 	return prometheus.UntypedValue
 }
 
-type writeRequest struct {
-	key string
-	val float64
-}
-
 type prometheusMetricDesc struct {
 	*prometheus.Desc
 	valueType prometheus.ValueType
@@ -46,112 +41,68 @@ type prometheusMetricDesc struct {
 	key       string
 }
 
-type prometheusExporter struct {
-	writerChan   chan *writeRequest
-	registerChan chan *prometheusMetricDesc
-	namespace    string
-	cancel       context.CancelFunc
+type PrometheusCollector struct {
+	namespace string
+	metrics   map[string]*prometheusMetricDesc
+	registry  *prometheus.Registry
 }
 
-func (pe *prometheusExporter) RegisterMetric(nm *MetricDescription) {
-	pe.registerChan <- &prometheusMetricDesc{
-		Desc:      prometheus.NewDesc(prometheus.BuildFQName(pe.namespace, "", nm.Name), nm.Help, nil, nm.ConstLabels),
-		valueType: convertValueType(nm.Type),
-		valptr:    new(uint64),
-		key:       nm.Key,
-	}
-}
-
-func (pe *prometheusExporter) UpdateMetric(key string, val float64) {
-	pe.writerChan <- &writeRequest{key: key, val: val}
-}
-
-func (pe *prometheusExporter) Shutdown() {
-	if pe.cancel != nil {
-		pe.cancel()
-	}
-}
-
-type prometheusCollector struct {
-	metrics map[string]*prometheusMetricDesc
-}
-
-func (pc *prometheusCollector) Describe(ch chan<- *prometheus.Desc) {
+func (pc *PrometheusCollector) Describe(ch chan<- *prometheus.Desc) {
 	for _, v := range pc.metrics {
+		logger.Println("collector describe :", v.Desc.String())
 		ch <- v.Desc
 	}
 }
 
-func (pc *prometheusCollector) Collect(ch chan<- prometheus.Metric) {
+func (pc *PrometheusCollector) Collect(ch chan<- prometheus.Metric) {
 	for _, v := range pc.metrics {
-		cm, err := prometheus.NewConstMetric(v.Desc, v.valueType, math.Float64frombits(atomic.LoadUint64(v.valptr)))
+		value := atomic.LoadUint64(v.valptr)
+		cm, err := prometheus.NewConstMetric(v.Desc, v.valueType, math.Float64frombits(value))
 		if err == nil {
 			ch <- cm
 		}
 	}
 }
 
-func (pe *prometheusExporter) loop(ctx context.Context) {
-	defer func() {
-		r := recover()
-		if r != nil {
-			logger.Error(r)
-		}
-	}()
+func (pc *PrometheusCollector) RegisterMetric(md *MetricDescription) *PrometheusCollector {
+	nm := &prometheusMetricDesc{
+		Desc:      prometheus.NewDesc(prometheus.BuildFQName("ou", "", md.Name), md.Help, nil, md.ConstLabels),
+		valueType: convertValueType(md.Type),
+		valptr:    new(uint64),
+		key:       md.Key,
+	}
 
-	var collector *prometheusCollector
-	defer func() {
-		if collector != nil {
-			prometheus.Unregister(collector)
-		}
-	}()
+	next := NewPrometheusCollector(pc.namespace, pc.registry)
+	maps.Copy(next.metrics, pc.metrics)
+	next.metrics[nm.key] = nm
 
-	for {
-		select {
-		case <-ctx.Done():
-			return
+	pc.registry.Unregister(pc)
+	pc.registry.Register(next)
 
-		case req := <-pe.writerChan:
-			if collector != nil {
-				if m := collector.metrics[req.key]; m != nil {
-					atomic.StoreUint64(m.valptr, math.Float64bits(req.val))
-				}
-			}
+	return next
+}
 
-		case nm := <-pe.registerChan:
-			var nextmetrics map[string]*prometheusMetricDesc
-			if collector != nil {
-				nextmetrics = collector.metrics
-				prometheus.Unregister(collector)
-				nextmetrics[nm.key] = nm
-			} else {
-				nextmetrics = map[string]*prometheusMetricDesc{
-					nm.key: nm,
-				}
-			}
-
-			nextcollector := &prometheusCollector{
-				metrics: nextmetrics,
-			}
-
-			if err := prometheus.Register(nextcollector); err != nil {
-				logger.Error("prometheus register err :", *nm, err)
-			} else {
-				collector = nextcollector
-			}
-		}
+func (pc *PrometheusCollector) UpdateMetric(key string, val float64) {
+	if m := pc.metrics[key]; m != nil {
+		atomic.StoreUint64(m.valptr, math.Float64bits(val))
 	}
 }
 
-func NewPrometheusExport(namespace string) Exporter {
-	ctx, cancel := context.WithCancel(context.Background())
-	exp := &prometheusExporter{
-		registerChan: make(chan *prometheusMetricDesc, 10),
-		writerChan:   make(chan *writeRequest, 100),
-		namespace:    namespace,
-		cancel:       cancel,
-	}
+func (pc *PrometheusCollector) UnregisterMetric(key string) *PrometheusCollector {
+	next := NewPrometheusCollector(pc.namespace, pc.registry)
+	maps.Copy(next.metrics, pc.metrics)
+	delete(next.metrics, key)
 
-	go exp.loop(ctx)
-	return exp
+	pc.registry.Unregister(pc)
+	pc.registry.Register(next)
+
+	return next
+}
+
+func NewPrometheusCollector(namespace string, registry *prometheus.Registry) *PrometheusCollector {
+	return &PrometheusCollector{
+		namespace: namespace,
+		metrics:   make(map[string]*prometheusMetricDesc),
+		registry:  registry,
+	}
 }

mongo.go

@@ -154,10 +154,36 @@ func (mc *MongoClient) DropIndex(coll CollectionName, name string) error {
 }
 
 func (mc *MongoClient) Watch(coll CollectionName, pipeline mongo.Pipeline, opts ...*options.ChangeStreamOptions) (*mongo.ChangeStream, error) {
+	// mc.db.RunCommand()
 	if len(opts) == 0 {
 		opts = []*options.ChangeStreamOptions{options.ChangeStream().SetFullDocument(options.UpdateLookup).SetMaxAwaitTime(0)}
 	}
-	return mc.Collection(coll).Watch(mc.ctx, pipeline, opts...)
+
+	stream, err := mc.Collection(coll).Watch(mc.ctx, pipeline, opts...)
+	if err != nil {
+		if mongoErr, ok := err.(mongo.CommandError); ok {
+			logger.Println("MongoClient Watch return err code :", mongoErr, mongoErr.Code)
+			if mongoErr.Code == 40573 {
+				adminDb := mc.db.Client().Database("admin")
+				result := adminDb.RunCommand(mc.ctx, bson.D{
+					{Key: "modifyChangeStreams", Value: 1},
+					{Key: "database", Value: mc.db.Name()},
+					{Key: "collection", Value: coll},
+					{Key: "enable", Value: true},
+				})
+
+				if result.Err() != nil {
+					logger.Println("mc.db.RunCommand failed :", result.Err(), mc.db.Name(), coll)
+				} else {
+					return mc.Collection(coll).Watch(mc.ctx, pipeline, opts...)
+				}
+			}
+		}
+
+		logger.Fatal(err)
+	}
+
+	return stream, err
 }
 
 func (mc *MongoClient) Collection(collname CollectionName) *mongo.Collection {

opensearch/client.go

@@ -0,0 +1,458 @@
+package opensearch
+
+import (
+	"context"
+	"crypto/hmac"
+	"crypto/sha256"
+	"encoding/base64"
+	"encoding/json"
+	"fmt"
+	"io"
+	"net"
+	"net/http"
+	"slices"
+	"strings"
+	"time"
+
+	osg "github.com/opensearch-project/opensearch-go/v4"
+	osapi "github.com/opensearch-project/opensearch-go/v4/opensearchapi"
+	"repositories.action2quare.com/ayo/gocommon/logger"
+)
+
+const logbulksize = 512 * 1024
+
+type Config struct {
+	osg.Config  `json:",inline"`
+	IndexPrefix string `json:"IndexPrefix"`
+	SigningKey  string `json:"SigningKey"`
+}
+
+type Client struct {
+	*osg.Client
+	cfg                  Config
+	signingKey           []byte
+	indexTemplatePattern string
+	bulkHeader           http.Header
+	singleHeader         http.Header
+	bulkChan             chan *LogDocument
+	singleLogPrepend     []byte
+	singleLogMidpend     []byte
+	singleLogAppend      []byte
+	singleLogFixedSize   int
+}
+
+type LogDocument struct {
+	Type      string `json:"type"`
+	Body      any    `json:"body"`
+	Timestamp string `json:"@timestamp"`
+
+	Country string `json:"country"`
+	Ip      string `json:"ip"`
+	Uid     string `json:"uid"`
+	Auth    struct {
+		Type string `json:"type"`
+		Id   string `json:"id"`
+	} `json:"auth"`
+}
+
+func NewLogDocument(logType string, body any) *LogDocument {
+	return &LogDocument{
+		Type:      strings.ToLower(logType),
+		Timestamp: time.Now().UTC().Format("2006-01-02T15:04:05Z"),
+		Body:      body,
+	}
+}
+
+func (c *Client) Send(ld *LogDocument) {
+	if c.Client == nil {
+		return
+	}
+
+	c.bulkChan <- ld
+}
+
+func (c *Client) SendBulk(ds map[string]*LogDocument) {
+	if c == nil {
+		return
+	}
+	for _, d := range ds {
+		c.bulkChan <- d
+	}
+}
+
+type singleLogMarshaller struct {
+	singleLogPrepend []byte
+	singleLogMidpend []byte
+	singleLogAppend  []byte
+
+	logtype []byte
+	content []byte
+	length  int
+}
+
+type logSliceReader struct {
+	src    []*singleLogMarshaller
+	cursor int
+}
+
+func newLogSliceReader(in []singleLogMarshaller) *logSliceReader {
+	src := make([]*singleLogMarshaller, len(in))
+	for i, v := range in {
+		copylog := new(singleLogMarshaller)
+		*copylog = v
+		src[i] = copylog
+	}
+
+	return &logSliceReader{
+		src:    src,
+		cursor: 0,
+	}
+}
+
+func (b *logSliceReader) Read(p []byte) (n int, err error) {
+	n = 0
+	err = nil
+
+	advance := func(in []byte) []byte {
+		if len(in) == 0 {
+			return in
+		}
+
+		copied := copy(p, in)
+		p = p[copied:]
+		n += copied
+		return in[copied:]
+	}
+
+	for b.cursor < len(b.src) {
+		sbt := b.src[b.cursor]
+
+		if sbt.singleLogPrepend = advance(sbt.singleLogPrepend); len(sbt.singleLogPrepend) > 0 {
+			return
+		}
+
+		if sbt.logtype = advance(sbt.logtype); len(sbt.logtype) > 0 {
+			return
+		}
+
+		if sbt.singleLogMidpend = advance(sbt.singleLogMidpend); len(sbt.singleLogMidpend) > 0 {
+			return
+		}
+
+		if sbt.content = advance(sbt.content); len(sbt.content) > 0 {
+			return
+		}
+
+		if sbt.singleLogAppend = advance(sbt.singleLogAppend); len(sbt.singleLogAppend) > 0 {
+			return
+		}
+
+		b.cursor++
+	}
+
+	err = io.EOF
+	return
+}
+
+func (b *logSliceReader) printSent() {
+	for _, r := range b.src {
+		fmt.Print(string(r.content))
+	}
+	fmt.Print("\n")
+}
+
+func (c *Client) sendLoop(ctx context.Context) {
+	defer func() {
+		r := recover()
+		if r != nil {
+			logger.Error(r)
+		}
+	}()
+
+	failChan := make(chan []singleLogMarshaller)
+	var logMarshallers []singleLogMarshaller
+	sendTick := time.After(time.Minute)
+
+	sendfunc := func(logs []singleLogMarshaller) {
+		if len(logs) == 0 {
+			return
+		}
+
+		defer func() {
+			r := recover()
+			if r != nil {
+				logger.Println(r)
+			}
+		}()
+
+		reader := newLogSliceReader(logs)
+		req := osapi.BulkReq{
+			Body:   reader,
+			Header: c.bulkHeader,
+		}
+		resp, err := c.Do(context.Background(), req, nil)
+
+		if err != nil {
+			if netoperr, ok := err.(*net.OpError); ok && netoperr.Op == "dial" {
+				// 접속 안됨. 재시도 안함
+				logger.Println("[LogStream] send bulk failed. no retry :", err)
+				reader.printSent()
+			} else {
+				// 재시도
+				logger.Println("[LogStream] send bulk failed. retry :", err)
+				failChan <- logs
+			}
+			return
+		}
+		if resp.Body == nil {
+			return
+		}
+		defer resp.Body.Close()
+
+		var respbody struct {
+			Errors bool `json:"errors"`
+			Items  []struct {
+				Create struct {
+					Status int `json:"status"`
+				} `json:"create"`
+			} `json:"items"`
+		}
+
+		decoder := json.NewDecoder(resp.Body)
+		if err := decoder.Decode(&respbody); err != nil {
+			errbody, _ := io.ReadAll(decoder.Buffered())
+			logger.Println("[LogStream] decode response body failed and retry :", err, string(errbody), len(logs))
+			// 전체 재시도 필요
+			failChan <- logs
+			return
+		}
+
+		if !respbody.Errors {
+			// 성공
+			return
+		}
+
+		var retry []singleLogMarshaller
+		for i, item := range respbody.Items {
+			if item.Create.Status < 300 {
+				continue
+			}
+
+			if item.Create.Status == 429 || item.Create.Status >= 500 {
+				logger.Println("[LogStream] send bulk failed but retry. status :", item.Create.Status)
+				retry = append(retry, logs[i])
+			} else if item.Create.Status == 400 {
+				// 구문 오류. 재시도 불가
+				if i < len(logs) {
+					logger.Println("[LogStream] send bulk failed. status 400 :", string(logs[i].content))
+				} else {
+					logger.Println("[LogStream] send bulk failed. status 400 but out of index :", i, len(logs))
+				}
+			} else {
+				// 일단 로그만
+				logger.Println("[LogStream] send bulk failed but no retry. status :", item.Create.Status)
+			}
+		}
+
+		if len(retry) > 0 {
+			failChan <- retry
+		}
+	}
+
+	totalsize := 0
+	appendLog := func(newlog singleLogMarshaller) bool {
+		if totalsize+newlog.length > logbulksize {
+			go sendfunc(logMarshallers)
+			totalsize = newlog.length
+			logMarshallers = []singleLogMarshaller{newlog}
+			return true
+		}
+
+		totalsize += newlog.length
+		logMarshallers = append(logMarshallers, newlog)
+		return false
+	}
+
+	for {
+		select {
+		case <-ctx.Done():
+			return
+
+		case ret := <-failChan:
+			// 순서는 중요하지 않음.
+			sent := false
+			for _, newlog := range ret {
+				sent = sent || appendLog(newlog)
+			}
+
+			if sent {
+				sendTick = time.After(time.Minute)
+			}
+
+		case <-sendTick:
+			if len(logMarshallers) > 0 {
+				go sendfunc(logMarshallers)
+				totalsize = 0
+				logMarshallers = nil
+			} else {
+				sendTick = time.After(time.Minute)
+			}
+
+		case logDoc := <-c.bulkChan:
+			b, _ := json.Marshal(logDoc)
+			logtype := []byte(logDoc.Type)
+
+			if appendLog(singleLogMarshaller{
+				singleLogPrepend: c.singleLogPrepend,
+				singleLogMidpend: c.singleLogMidpend,
+				singleLogAppend:  c.singleLogAppend,
+				logtype:          logtype,
+				content:          b,
+				length:           len(logtype) + len(b) + c.singleLogFixedSize,
+			}) {
+				sendTick = time.After(time.Minute)
+			}
+		}
+	}
+}
+
+var jwtHeader string
+var encoding = base64.RawURLEncoding
+
+func init() {
+	src := []byte(`{"alg": "HS256","typ": "JWT"}`)
+	dst := make([]byte, len(src)*2)
+	encoding.Encode(dst, src)
+	enclen := encoding.EncodedLen(len(src))
+	jwtHeader = string(dst[:enclen])
+}
+
+func (c *Client) MakeJWT(subject string, role string, ttl time.Duration) string {
+	if len(c.signingKey) == 0 {
+		return ""
+	}
+
+	now := time.Now().Add(ttl).Unix()
+	src := fmt.Appendf(nil, `{"exp":%d,"sub":"%s","roles":"%s"}`, now, subject, role)
+	payload := make([]byte, encoding.EncodedLen(len(src)))
+	encoding.Encode(payload, src)
+
+	encoded := jwtHeader + "." + string(payload)
+	mac := hmac.New(sha256.New, c.signingKey)
+	mac.Write([]byte(encoded))
+	signature := mac.Sum(nil)
+	sigenc := make([]byte, encoding.EncodedLen(len(signature)))
+	encoding.Encode(sigenc, signature)
+
+	return encoded + "." + string(sigenc)
+}
+
+func (c *Client) VerifyJWT(token string) (subject string, role string) {
+	dot := strings.LastIndex(token, ".")
+	if dot < 0 {
+		return
+	}
+
+	encoded := token[:dot]
+	sigenc := token[dot+1:]
+	signature := make([]byte, encoding.DecodedLen(len(sigenc)))
+	encoding.Decode(signature, []byte(sigenc))
+
+	mac := hmac.New(sha256.New, c.signingKey)
+	mac.Write([]byte(encoded))
+	calsig := mac.Sum(nil)
+	if slices.Compare(calsig, signature) != 0 {
+		return
+	}
+
+	_, payload, ok := strings.Cut(encoded, ".")
+	if !ok {
+		return
+	}
+
+	srcjson, err := encoding.DecodeString(payload)
+	if err != nil {
+		return
+	}
+
+	var src struct {
+		Exp   int64  `json:"exp"`
+		Sub   string `json:"sub"`
+		Roles string `json:"roles"`
+	}
+	if json.Unmarshal([]byte(srcjson), &src) != nil {
+		return
+	}
+	if src.Exp < time.Now().Unix() {
+		return
+	}
+
+	return src.Sub, src.Roles
+}
+
+func NewClient(ctx context.Context, cfg Config) (Client, error) {
+	if len(cfg.Addresses) == 0 {
+		return Client{}, nil
+	}
+
+	// retry는 수동으로
+	cfg.Config.DisableRetry = true
+	client, err := osg.NewClient(cfg.Config)
+	if err != nil {
+		return Client{}, err
+	}
+
+	var signingKey []byte
+	if len(cfg.SigningKey) > 0 {
+		dst := make([]byte, len(cfg.SigningKey)*2)
+		dstlen, _ := base64.StdEncoding.Decode(dst, []byte(cfg.SigningKey))
+		signingKey = dst[:dstlen]
+	}
+
+	indexPrefix := cfg.IndexPrefix
+	if !strings.HasSuffix(indexPrefix, "-") && len(indexPrefix) > 0 {
+		indexPrefix += "-"
+	}
+	if !strings.HasSuffix(indexPrefix, "ds-logs-") {
+		indexPrefix = "ds-logs-" + indexPrefix
+	}
+
+	logger.Println("[LogStream] stream indexPrefix :", indexPrefix)
+	bulkHeader := make(http.Header)
+	singleHeader := make(http.Header)
+	if len(cfg.Username) > 0 && len(cfg.Password) > 0 {
+		authHeader := fmt.Sprintf("Basic %s", base64.RawURLEncoding.EncodeToString(fmt.Appendf(nil, "%s:%s", cfg.Username, cfg.Password)))
+		bulkHeader.Set("Authorization", authHeader)
+		singleHeader.Set("Authorization", authHeader)
+	}
+
+	singleLogPrepend := fmt.Appendf(nil, `{"create":{"_index":"%s`, indexPrefix)
+	singleLogMidpend := []byte("\"}}\n")
+	singleLogAppend := []byte("\n")
+	singleLogFixedSize := len(singleLogPrepend) + len(singleLogMidpend) + len(singleLogAppend)
+
+	out := Client{
+		Client:               client,
+		cfg:                  cfg,
+		signingKey:           signingKey,
+		indexTemplatePattern: indexPrefix,
+		bulkHeader:           bulkHeader,
+		singleHeader:         singleHeader,
+		bulkChan:             make(chan *LogDocument, 1000),
+
+		singleLogPrepend:   singleLogPrepend,
+		singleLogMidpend:   singleLogMidpend,
+		singleLogAppend:    singleLogAppend,
+		singleLogFixedSize: singleLogFixedSize,
+	}
+
+	go func() {
+		for {
+			out.sendLoop(ctx)
+			if ctx.Err() != nil {
+				return
+			}
+		}
+	}()
+	return out, nil
+}

opensearch/client_test.go

@@ -0,0 +1,78 @@
+package opensearch
+
+import (
+	"encoding/base64"
+	"testing"
+	"time"
+
+	"go.mongodb.org/mongo-driver/bson/primitive"
+)
+
+func TestNewClient(t *testing.T) {
+	// var cfg Config
+
+	// cfg.Addresses = []string{"http://localhost:9200"}
+	// client, err := NewClient(cfg)
+	// if err != nil {
+	// 	t.Errorf("NewClient() error = %v", err)
+	// 	return
+	// }
+
+	// for i := 0; i < 10; i++ {
+	// 	MakeActorLog("LOGIN", "test_user", "stalkername").Write(&client, bson.M{
+	// 		"country": "kr",
+	// 		"ip":      "127.0.0.1",
+	// 	})
+	// 	time.Sleep(time.Second)
+	// }
+
+	// for i := 0; i < 10; i++ {
+	// 	MakeActorLog("Match", "test_user", "stalkername").Write(&client, bson.M{
+	// 		"server":  "kr001",
+	// 		"mode":    "pvp",
+	// 		"address": "server address",
+	// 	})
+	// 	time.Sleep(time.Second)
+	// }
+}
+
+func TestClient_MakeJWT(t *testing.T) {
+	sk := "UGdiOTdLVjFBTWtndTRNRiZmVjdwMDdCRW1lSSUxTnA="
+	dst := make([]byte, len(sk)*2)
+	dstlen, _ := base64.StdEncoding.Decode(dst, []byte(sk))
+	signingKey := dst[:dstlen]
+	uid := primitive.NewObjectID().Hex()
+
+	type args struct {
+		subject string
+		role    string
+		ttl     time.Duration
+	}
+	tests := []struct {
+		name string
+		c    *Client
+		args args
+		want string
+	}{
+		// TODO: Add test cases.
+		{
+			c: &Client{
+				signingKey: signingKey,
+			},
+			args: args{
+				subject: uid,
+				role:    "ds_client",
+				ttl:     time.Minute,
+			},
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			got := tt.c.MakeJWT(tt.args.subject, tt.args.role, tt.args.ttl)
+			subj, role := tt.c.VerifyJWT(got)
+			if subj != tt.args.subject || role != tt.args.role {
+				t.Errorf("Client.MakeJWT() = %v, %v, want %v, %v", subj, role, tt.args.subject, tt.args.role)
+			}
+		})
+	}
+}

reflect_config.go

@@ -116,7 +116,13 @@ func LoadConfig[T any](outptr *T) error {
 		}
 	}
 
-	return json.Unmarshal([]byte(os.ExpandEnv(string(configContents))), outptr)
+	return json.Unmarshal([]byte(os.Expand(string(configContents), func(in string) string {
+		envval := os.Getenv(in)
+		if len(envval) == 0 {
+			return "$" + in
+		}
+		return envval
+	})), outptr)
 }
 
 type StorageAddr struct {

server.go

@@ -14,6 +14,7 @@ import (
 	"net/url"
 	"os"
 	"os/signal"
+	"path"
 	"reflect"
 	"runtime"
 	"strconv"
@@ -40,6 +41,13 @@ func init() {
 	gob.Register([]any{})
 }
 
+type ServerMuxInterface interface {
+	http.Handler
+
+	HandleFunc(pattern string, handler func(http.ResponseWriter, *http.Request))
+	Handle(pattern string, handler http.Handler)
+}
+
 const (
 	// HTTPStatusReloginRequired : http status를 이걸 받으면 클라이언트는 로그아웃하고 로그인 화면으로 돌아가야 한다.
 	HTTPStatusReloginRequired = 599
@@ -139,8 +147,19 @@ func isTlsEnabled(fileout ...*string) bool {
 	return true
 }
 
+func registUnhandledPattern(serveMux ServerMuxInterface) {
+	defer func() {
+		recover()
+	}()
+
+	serveMux.HandleFunc("/", func(w http.ResponseWriter, r *http.Request) {
+		logger.Println("page not found :", r.URL.Path)
+		w.WriteHeader(http.StatusNotFound)
+	})
+}
+
 // NewHTTPServer :
-func NewHTTPServerWithPort(serveMux *http.ServeMux, port int) *Server {
+func NewHTTPServerWithPort(serveMux ServerMuxInterface, port int) *Server {
 	if isTlsEnabled() && port == 80 {
 		port = 443
 	}
@@ -148,16 +167,21 @@ func NewHTTPServerWithPort(serveMux *http.ServeMux, port int) *Server {
 	serveMux.HandleFunc(MakeHttpHandlerPattern("welcome"), welcomeHandler)
 	serveMux.HandleFunc(MakeHttpHandlerPattern("lb_health_chceck"), healthCheckHandler)
 	serveMux.HandleFunc(MakeHttpHandlerPattern("lb_health_check"), healthCheckHandler)
+	registUnhandledPattern(serveMux)
 
 	server := &Server{
-		httpserver: &http.Server{Addr: addr, Handler: serveMux},
+		httpserver: &http.Server{
+			Addr:           addr,
+			Handler:        serveMux,
+			MaxHeaderBytes: 2 << 20, // 2 MB
+		},
 	}
 	server.httpserver.SetKeepAlivesEnabled(true)
 
 	return server
 }
 
-func NewHTTPServer(serveMux *http.ServeMux) *Server {
+func NewHTTPServer(serveMux ServerMuxInterface) *Server {
 
 	// 시작시 자동으로 enable됨
 	if isTlsEnabled() && *portptr == 80 {
@@ -206,7 +230,12 @@ func (server *Server) Stop() {
 }
 
 // Start :
-func (server *Server) Start() error {
+func (server *Server) Start(name ...string) error {
+	if len(name) == 0 {
+		exepath, _ := os.Executable()
+		name = []string{path.Base(exepath)}
+	}
+
 	if server.httpserver != nil {
 		ln, r := net.Listen("tcp", server.httpserver.Addr)
 		if r != nil {
@@ -235,6 +264,7 @@ func (server *Server) Start() error {
 			err = server.httpserver.ServeTLS(proxyListener, crtfile, keyfile)
 		} else {
 			logger.Println("tls disabled")
+			logger.Println(strings.Join(name, ", "), "started")
 			err = server.httpserver.Serve(proxyListener)
 		}
 
@@ -322,8 +352,19 @@ func ConvertInterface(from interface{}, toType reflect.Type) reflect.Value {
 		return convslice
 
 	case reflect.Bool:
-		val, _ := strconv.ParseBool(from.(string))
-		return reflect.ValueOf(val)
+		if fromstr, ok := from.(string); ok {
+			val, _ := strconv.ParseBool(fromstr)
+			return reflect.ValueOf(val)
+		} else if frombool, ok := from.(bool); ok {
+			return reflect.ValueOf(frombool)
+		}
+		return reflect.ValueOf(false)
+
+	case reflect.String:
+		if toType == reflect.TypeOf(primitive.ObjectID{}) {
+			objid, _ := primitive.ObjectIDFromHex(from.(string))
+			return reflect.ValueOf(objid)
+		}
 	}
 
 	return fromrv.Convert(toType)
@@ -481,6 +522,13 @@ func ReadStringFormValue(r url.Values, key string) (string, bool) {
 	return strval, len(strval) > 0
 }
 
+func ReadStringsFormValue(r url.Values, key string) ([]string, bool) {
+	if r.Has(key) {
+		return (map[string][]string)(r)[key], true
+	}
+	return nil, false
+}
+
 type encoder interface {
 	Encode(any) error
 }

session/common.go

@@ -6,6 +6,7 @@ import (
 	"encoding/hex"
 	"errors"
 	"math/rand"
+	"strconv"
 	"strings"
 	"time"
 
@@ -13,13 +14,13 @@ import (
 )
 
 type Authorization struct {
-	Account     primitive.ObjectID `bson:"a" json:"a"`
-	invalidated string
+	Account primitive.ObjectID `bson:"a" json:"a"`
 
 	// by authorization provider
-	Platform string `bson:"p" json:"p"`
-	Uid      string `bson:"u" json:"u"`
-	Email    string `bson:"em" json:"em"`
+	Platform    string `bson:"p" json:"p"`
+	Uid         string `bson:"u" json:"u"`
+	Alias       string `bson:"al" json:"al"`
+	CreatedTime int64  `bson:"ct" json:"ct"`
 }
 
 func (auth *Authorization) ToStrings() []string {
@@ -27,39 +28,46 @@ func (auth *Authorization) ToStrings() []string {
 		"a", auth.Account.Hex(),
 		"p", auth.Platform,
 		"u", auth.Uid,
-		"em", auth.Email,
-		"inv", auth.invalidated,
+		"al", auth.Alias,
+		"ct", strconv.FormatInt(auth.CreatedTime, 10),
 	}
 }
 
-func (auth *Authorization) Invalidated() bool {
-	return len(auth.invalidated) > 0
+func (auth *Authorization) Valid() bool {
+	return !auth.Account.IsZero()
 }
 
 func MakeAuthrizationFromStringMap(src map[string]string) Authorization {
 	accid, _ := primitive.ObjectIDFromHex(src["a"])
+	ct, _ := strconv.ParseInt(src["ct"], 10, 0)
 	return Authorization{
 		Account:     accid,
 		Platform:    src["p"],
 		Uid:         src["u"],
-		Email:       src["em"],
-		invalidated: src["inv"],
+		Alias:       src["al"],
+		CreatedTime: ct,
 	}
 }
 
 type Provider interface {
 	New(*Authorization) (string, error)
-	RevokeAll(primitive.ObjectID) error
+	RevokeAll(primitive.ObjectID, bool) ([]string, error)
 	Query(string) (Authorization, error)
 	Touch(string) (bool, error)
 }
 
+type InvalidatedSession struct {
+	Account     primitive.ObjectID
+	SessionKeys []string
+	Infinite    bool
+}
+
 type Consumer interface {
-	Query(string) (Authorization, error)
+	Query(string) Authorization
 	Touch(string) (Authorization, error)
 	IsRevoked(primitive.ObjectID) bool
 	Revoke(string)
-	RegisterOnSessionInvalidated(func(primitive.ObjectID))
+	RegisterOnSessionInvalidated(func(InvalidatedSession))
 }
 
 type storagekey string
@@ -74,10 +82,6 @@ func make_storagekey(acc primitive.ObjectID) storagekey {
 	return storagekey(acc.Hex() + hex.EncodeToString(bs[2:]))
 }
 
-func AccountToSessionKey(acc primitive.ObjectID) string {
-	return string(make_storagekey(acc))
-}
-
 func storagekey_to_publickey(sk storagekey) publickey {
 	bs, _ := hex.DecodeString(string(sk))
 
@@ -119,10 +123,6 @@ var errInvalidScheme = errors.New("storageAddr is not valid scheme")
 var errSessionStorageMissing = errors.New("session_storageis missing")
 
 func NewConsumer(ctx context.Context, storageAddr string, ttl time.Duration) (Consumer, error) {
-	if strings.HasPrefix(storageAddr, "mongodb") {
-		return newConsumerWithMongo(ctx, storageAddr, ttl)
-	}
-
 	if strings.HasPrefix(storageAddr, "redis") {
 		return newConsumerWithRedis(ctx, storageAddr, ttl)
 	}
@@ -142,10 +142,6 @@ func NewConsumerWithConfig(ctx context.Context, cfg SessionConfig) (Consumer, er
 }
 
 func NewProvider(ctx context.Context, storageAddr string, ttl time.Duration) (Provider, error) {
-	if strings.HasPrefix(storageAddr, "mongodb") {
-		return newProviderWithMongo(ctx, storageAddr, ttl)
-	}
-
 	if strings.HasPrefix(storageAddr, "redis") {
 		return newProviderWithRedis(ctx, storageAddr, ttl)
 	}

session/consumer_common.go

@@ -4,8 +4,6 @@ import (
 	"context"
 	"sync"
 	"time"
-
-	"go.mongodb.org/mongo-driver/bson/primitive"
 )
 
 type cache_stage[T any] struct {
@@ -26,7 +24,7 @@ type consumer_common[T any] struct {
 	ctx                  context.Context
 	stages               [2]*cache_stage[T]
 	startTime            time.Time
-	onSessionInvalidated []func(primitive.ObjectID)
+	onSessionInvalidated []func(InvalidatedSession)
 }
 
 func (c *consumer_common[T]) add_internal(sk storagekey, si T) {

session/impl_mongo.go

@@ -1,383 +0,0 @@
-package session
-
-import (
-	"context"
-	"time"
-
-	"go.mongodb.org/mongo-driver/bson"
-	"go.mongodb.org/mongo-driver/bson/primitive"
-	"go.mongodb.org/mongo-driver/mongo"
-	"go.mongodb.org/mongo-driver/mongo/options"
-	"repositories.action2quare.com/ayo/gocommon"
-	"repositories.action2quare.com/ayo/gocommon/logger"
-)
-
-const (
-	session_collection_name = gocommon.CollectionName("session")
-)
-
-type provider_mongo struct {
-	mongoClient gocommon.MongoClient
-}
-
-type sessionMongo struct {
-	Id   primitive.ObjectID `bson:"_id,omitempty"`
-	Auth *Authorization     `bson:"auth"`
-	Key  storagekey         `bson:"key"`
-	Ts   primitive.DateTime `bson:"_ts"`
-}
-
-func newProviderWithMongo(ctx context.Context, mongoUrl string, ttl time.Duration) (Provider, error) {
-	mc, err := gocommon.NewMongoClient(ctx, mongoUrl)
-	if err != nil {
-		return nil, err
-	}
-
-	if err = mc.MakeUniqueIndices(session_collection_name, map[string]bson.D{
-		"key": {{Key: "key", Value: 1}},
-	}); err != nil {
-		return nil, err
-	}
-
-	if err := mc.MakeExpireIndex(session_collection_name, int32(ttl.Seconds())); err != nil {
-		return nil, err
-	}
-
-	return &provider_mongo{
-		mongoClient: mc,
-	}, nil
-}
-
-func (p *provider_mongo) New(input *Authorization) (string, error) {
-	sk := make_storagekey(input.Account)
-
-	_, _, err := p.mongoClient.Update(session_collection_name, bson.M{
-		"_id": input.Account,
-	}, bson.M{
-		"$set": sessionMongo{
-			Auth: input,
-			Key:  sk,
-			Ts:   primitive.NewDateTimeFromTime(time.Now().UTC()),
-		},
-	}, options.Update().SetUpsert(true))
-
-	return string(storagekey_to_publickey(sk)), err
-}
-
-func (p *provider_mongo) RevokeAll(acc primitive.ObjectID) error {
-	_, err := p.mongoClient.Delete(session_collection_name, bson.M{
-		"_id": acc,
-	})
-	return err
-}
-
-func (p *provider_mongo) Query(pk string) (Authorization, error) {
-	sk := publickey_to_storagekey(publickey(pk))
-	var auth Authorization
-	err := p.mongoClient.FindOneAs(session_collection_name, bson.M{
-		"key": sk,
-	}, &auth)
-
-	return auth, err
-}
-
-func (p *provider_mongo) Touch(pk string) (bool, error) {
-	sk := publickey_to_storagekey(publickey(pk))
-	worked, _, err := p.mongoClient.Update(session_collection_name, bson.M{
-		"key": sk,
-	}, bson.M{
-		"$currentDate": bson.M{
-			"_ts": bson.M{"$type": "date"},
-		},
-	}, options.Update().SetUpsert(false))
-
-	if err != nil {
-		logger.Println("provider Touch :", err)
-		return false, err
-	}
-
-	return worked, nil
-}
-
-type consumer_mongo struct {
-	consumer_common[*sessionMongo]
-	ids         map[primitive.ObjectID]storagekey
-	mongoClient gocommon.MongoClient
-	ttl         time.Duration
-}
-
-type sessionPipelineDocument struct {
-	OperationType string `bson:"operationType"`
-	DocumentKey   struct {
-		Id primitive.ObjectID `bson:"_id"`
-	} `bson:"documentKey"`
-	Session *sessionMongo `bson:"fullDocument"`
-}
-
-func newConsumerWithMongo(ctx context.Context, mongoUrl string, ttl time.Duration) (Consumer, error) {
-	mc, err := gocommon.NewMongoClient(ctx, mongoUrl)
-	if err != nil {
-		return nil, err
-	}
-
-	consumer := &consumer_mongo{
-		consumer_common: consumer_common[*sessionMongo]{
-			ttl:       ttl,
-			ctx:       ctx,
-			stages:    [2]*cache_stage[*sessionMongo]{make_cache_stage[*sessionMongo](), make_cache_stage[*sessionMongo]()},
-			startTime: time.Now(),
-		},
-		ids:         make(map[primitive.ObjectID]storagekey),
-		ttl:         ttl,
-		mongoClient: mc,
-	}
-
-	go func() {
-		matchStage := bson.D{
-			{
-				Key: "$match", Value: bson.D{
-					{Key: "operationType", Value: bson.D{
-						{Key: "$in", Value: bson.A{
-							"delete",
-							"insert",
-							"update",
-						}},
-					}},
-				},
-			}}
-		projectStage := bson.D{
-			{
-				Key: "$project", Value: bson.D{
-					{Key: "documentKey", Value: 1},
-					{Key: "operationType", Value: 1},
-					{Key: "fullDocument", Value: 1},
-				},
-			},
-		}
-
-		var stream *mongo.ChangeStream
-		nextswitch := time.Now().Add(ttl)
-		for {
-			if stream == nil {
-				stream, err = mc.Watch(session_collection_name, mongo.Pipeline{matchStage, projectStage})
-				if err != nil {
-					logger.Error("watchAuthCollection watch failed :", err)
-					time.Sleep(time.Minute)
-					continue
-				}
-			}
-
-			changed := stream.TryNext(ctx)
-			if ctx.Err() != nil {
-				logger.Error("watchAuthCollection stream.TryNext failed. process should be restarted! :", ctx.Err().Error())
-				break
-			}
-
-			if changed {
-				var data sessionPipelineDocument
-				if err := stream.Decode(&data); err == nil {
-					ot := data.OperationType
-					switch ot {
-					case "insert":
-						consumer.add(data.Session.Key, data.DocumentKey.Id, data.Session)
-					case "update":
-						if data.Session == nil {
-							if old := consumer.deleteById(data.DocumentKey.Id); old != nil {
-								for _, f := range consumer.onSessionInvalidated {
-									f(old.Auth.Account)
-								}
-							}
-						} else {
-							consumer.add(data.Session.Key, data.DocumentKey.Id, data.Session)
-						}
-					case "delete":
-						if old := consumer.deleteById(data.DocumentKey.Id); old != nil {
-							for _, f := range consumer.onSessionInvalidated {
-								f(old.Auth.Account)
-							}
-						}
-					}
-				} else {
-					logger.Error("watchAuthCollection stream.Decode failed :", err)
-				}
-			} else if stream.Err() != nil || stream.ID() == 0 {
-				select {
-				case <-ctx.Done():
-					logger.Println("watchAuthCollection is done")
-					stream.Close(ctx)
-					return
-
-				case <-time.After(time.Second):
-					logger.Error("watchAuthCollection stream error :", stream.Err())
-					stream.Close(ctx)
-					stream = nil
-				}
-			} else {
-				time.Sleep(time.Second)
-			}
-
-			now := time.Now()
-			for now.After(nextswitch) {
-				consumer.changeStage()
-				nextswitch = nextswitch.Add(ttl)
-			}
-		}
-	}()
-
-	return consumer, nil
-}
-
-func (c *consumer_mongo) query_internal(sk storagekey) (*sessionMongo, bool, error) {
-	if _, deleted := c.stages[0].deleted[sk]; deleted {
-		return nil, false, nil
-	}
-
-	if _, deleted := c.stages[1].deleted[sk]; deleted {
-		return nil, false, nil
-	}
-
-	found, ok := c.stages[0].cache[sk]
-	if !ok {
-		found, ok = c.stages[1].cache[sk]
-	}
-
-	if ok {
-		return found, false, nil
-	}
-
-	var si sessionMongo
-	err := c.mongoClient.FindOneAs(session_collection_name, bson.M{
-		"key": sk,
-	}, &si)
-
-	if err != nil {
-		logger.Println("consumer Query :", err)
-		return nil, false, err
-	}
-
-	if len(si.Key) > 0 {
-		siptr := &si
-		c.add_internal(sk, siptr)
-		return siptr, true, nil
-	}
-	return nil, false, nil
-}
-
-func (c *consumer_mongo) Query(pk string) (Authorization, error) {
-	c.lock.Lock()
-	defer c.lock.Unlock()
-
-	sk := publickey_to_storagekey(publickey(pk))
-	si, _, err := c.query_internal(sk)
-	if err != nil {
-		return Authorization{}, err
-	}
-
-	if si == nil {
-		return Authorization{}, nil
-	}
-
-	if time.Now().After(si.Ts.Time().Add(c.ttl)) {
-		return Authorization{}, nil
-	}
-
-	return *si.Auth, nil
-}
-
-func (c *consumer_mongo) Touch(pk string) (Authorization, error) {
-	c.lock.Lock()
-	defer c.lock.Unlock()
-
-	sk := publickey_to_storagekey(publickey(pk))
-	worked, _, err := c.mongoClient.Update(session_collection_name, bson.M{
-		"key": sk,
-	}, bson.M{
-		"$currentDate": bson.M{
-			"_ts": bson.M{"$type": "date"},
-		},
-	}, options.Update().SetUpsert(false))
-
-	if err != nil {
-		logger.Println("consumer Touch :", err)
-		return Authorization{}, err
-	}
-
-	if !worked {
-		// 이미 만료되서 사라짐
-		return Authorization{}, nil
-	}
-
-	si, added, err := c.query_internal(sk)
-	if err != nil {
-		return Authorization{}, err
-	}
-
-	if si == nil {
-		return Authorization{}, nil
-	}
-
-	if !added {
-		var doc sessionMongo
-		err := c.mongoClient.FindOneAs(session_collection_name, bson.M{
-			"key": sk,
-		}, &doc)
-
-		if err != nil {
-			logger.Println("consumer Query :", err)
-			return Authorization{}, err
-		}
-
-		if len(si.Key) > 0 {
-			c.add_internal(sk, &doc)
-			c.ids[doc.Id] = sk
-
-			return *doc.Auth, nil
-		}
-	}
-
-	return *si.Auth, nil
-}
-
-func (c *consumer_mongo) Revoke(pk string) {
-	sk := publickey_to_storagekey(publickey(pk))
-	_, err := c.mongoClient.Delete(session_collection_name, bson.M{
-		"key": sk,
-	})
-
-	if err == nil {
-		for id, v := range c.ids {
-			if v == sk {
-				delete(c.ids, id)
-				break
-			}
-		}
-	}
-}
-
-func (c *consumer_mongo) IsRevoked(id primitive.ObjectID) bool {
-	_, ok := c.ids[id]
-	return !ok
-}
-
-func (c *consumer_mongo) add(sk storagekey, id primitive.ObjectID, si *sessionMongo) {
-	c.lock.Lock()
-	defer c.lock.Unlock()
-
-	c.consumer_common.add_internal(sk, si)
-	c.ids[id] = sk
-}
-
-func (c *consumer_mongo) deleteById(id primitive.ObjectID) (old *sessionMongo) {
-	c.lock.Lock()
-	defer c.lock.Unlock()
-
-	if sk, ok := c.ids[id]; ok {
-		old = c.consumer_common.delete_internal(sk)
-		delete(c.ids, id)
-	}
-	return
-}
-
-func (c *consumer_mongo) RegisterOnSessionInvalidated(cb func(primitive.ObjectID)) {
-	c.onSessionInvalidated = append(c.onSessionInvalidated, cb)
-}

session/impl_redis.go

@@ -2,7 +2,10 @@ package session
 
 import (
 	"context"
+	"encoding/json"
+	"errors"
 	"fmt"
+	"slices"
 	"time"
 
 	"github.com/go-redis/redis/v8"
@@ -42,31 +45,18 @@ func newProviderWithRedis(ctx context.Context, redisUrl string, ttl time.Duratio
 }
 
 func (p *provider_redis) New(input *Authorization) (string, error) {
-	newsk := make_storagekey(input.Account)
-	prefix := input.Account.Hex()
-	sks, err := p.redisClient.Keys(p.ctx, prefix+"*").Result()
+	sks, err := p.RevokeAll(input.Account, false)
 	if err != nil {
-		logger.Println("session provider delete :", sks, err)
 		return "", err
 	}
 
-	p.redisClient.Del(p.ctx, sks...)
-	for _, sk := range sks {
-		p.redisClient.Publish(p.ctx, p.deleteChannel, sk).Result()
-	}
-
+	var newsk storagekey
 	for {
-		duplicated := false
-		for _, sk := range sks {
-			if sk == string(newsk) {
-				duplicated = true
-				break
-			}
-		}
+		newsk = make_storagekey(input.Account)
+		duplicated := slices.Contains(sks, string(newsk))
 		if !duplicated {
 			break
 		}
-		newsk = make_storagekey(input.Account)
 	}
 
 	_, err = p.redisClient.HSet(p.ctx, string(newsk), input.ToStrings()).Result()
@@ -81,20 +71,28 @@ func (p *provider_redis) New(input *Authorization) (string, error) {
 	return string(pk), err
 }
 
-func (p *provider_redis) RevokeAll(account primitive.ObjectID) error {
+func (p *provider_redis) RevokeAll(account primitive.ObjectID, infinite bool) ([]string, error) {
 	prefix := account.Hex()
 	sks, err := p.redisClient.Keys(p.ctx, prefix+"*").Result()
 	if err != nil {
 		logger.Println("session provider delete :", sks, err)
-		return err
+		return nil, err
 	}
 
-	for _, sk := range sks {
-		p.redisClient.HSet(p.ctx, sk, "inv", "true")
-		p.redisClient.Publish(p.ctx, p.deleteChannel, sk).Result()
+	if len(sks) > 0 {
+		p.redisClient.Del(p.ctx, sks...)
+
+		invsess := InvalidatedSession{
+			SessionKeys: sks,
+			Account:     account,
+			Infinite:    infinite,
+		}
+		data, _ := json.Marshal(invsess)
+
+		p.redisClient.Publish(p.ctx, p.deleteChannel, string(data)).Result()
 	}
 
-	return nil
+	return sks, nil
 }
 
 func (p *provider_redis) Query(pk string) (Authorization, error) {
@@ -180,12 +178,18 @@ func newConsumerWithRedis(ctx context.Context, redisUrl string, ttl time.Duratio
 
 				switch msg.Channel {
 				case deleteChannel:
-					sk := storagekey(msg.Payload)
-					old := consumer.delete(sk)
-					if old != nil {
-						for _, f := range consumer.onSessionInvalidated {
-							f(old.Account)
-						}
+					var invsess InvalidatedSession
+					if err := json.Unmarshal([]byte(msg.Payload), &invsess); err != nil {
+						logger.Println("redis consumer deleteChannel unmarshal failed :", err)
+						break
+					}
+
+					for _, sk := range invsess.SessionKeys {
+						consumer.delete(storagekey(sk))
+					}
+
+					for _, f := range consumer.onSessionInvalidated {
+						f(invsess)
 					}
 				}
 			}
@@ -243,37 +247,49 @@ func (c *consumer_redis) query_internal(sk storagekey) (*sessionRedis, error) {
 		expireAt:      time.Now().Add(ttl),
 	}
 
-	if auth.Invalidated() {
-		c.stages[0].deleted[sk] = si
-	} else {
+	if auth.Valid() {
 		c.add_internal(sk, si)
+	} else {
+		c.stages[0].deleted[sk] = si
 	}
 
 	return si, nil
 }
 
-func (c *consumer_redis) Query(pk string) (Authorization, error) {
+var errRevoked = errors.New("session revoked")
+var errExpired = errors.New("session expired")
+
+func (c *consumer_redis) Query(pk string) Authorization {
 	c.lock.Lock()
 	defer c.lock.Unlock()
 
 	sk := publickey_to_storagekey(publickey(pk))
+
+	if _, deleted := c.stages[0].deleted[sk]; deleted {
+		return Authorization{}
+	}
+
+	if _, deleted := c.stages[1].deleted[sk]; deleted {
+		return Authorization{}
+	}
+
 	si, err := c.query_internal(sk)
 	if err != nil {
 		logger.Println("session consumer query :", pk, err)
-		return Authorization{}, err
+		return Authorization{}
 	}
 
 	if si == nil {
 		logger.Println("session consumer query(si nil) :", pk, nil)
-		return Authorization{}, nil
+		return Authorization{}
 	}
 
 	if time.Now().After(si.expireAt) {
 		logger.Println("session consumer query(expired):", pk, nil)
-		return Authorization{}, nil
+		return Authorization{}
 	}
 
-	return *si.Authorization, nil
+	return *si.Authorization
 }
 
 func (c *consumer_redis) Touch(pk string) (Authorization, error) {
@@ -353,6 +369,6 @@ func (c *consumer_redis) IsRevoked(accid primitive.ObjectID) bool {
 	return false
 }
 
-func (c *consumer_redis) RegisterOnSessionInvalidated(cb func(primitive.ObjectID)) {
+func (c *consumer_redis) RegisterOnSessionInvalidated(cb func(InvalidatedSession)) {
 	c.onSessionInvalidated = append(c.onSessionInvalidated, cb)
 }

session/session_test.go

@@ -60,11 +60,11 @@ func TestExpTable(t *testing.T) {
 
 	go func() {
 		for {
-			q1, err := cs.Query(sk1)
-			logger.Println("query :", q1, err)
+			q1 := cs.Query(sk1)
+			logger.Println("query :", q1)
 
-			q2, err := cs.Query(sk2)
-			logger.Println("query :", q2, err)
+			q2 := cs.Query(sk2)
+			logger.Println("query :", q2)
 			time.Sleep(time.Second)
 		}
 	}()
@@ -75,7 +75,7 @@ func TestExpTable(t *testing.T) {
 	time.Sleep(2 * time.Second)
 
 	time.Sleep(2 * time.Second)
-	pv.RevokeAll(au1.Account)
+	pv.RevokeAll(au1.Account, false)
 
 	cs.Touch(sk1)
 	time.Sleep(2 * time.Second)
@@ -87,7 +87,7 @@ func TestExpTable(t *testing.T) {
 		t.Error(err)
 	}
 
-	q2, err := cs2.Query(sk2)
-	logger.Println("queryf :", q2, err)
+	q2 := cs2.Query(sk2)
+	logger.Println("queryf :", q2)
 	time.Sleep(20 * time.Second)
 }

wshandler/wshandler.go

@@ -188,6 +188,7 @@ type websocketHandlerBase struct {
 type WebsocketHandler struct {
 	WebsocketApiBroker
 	websocketHandlerBase
+	opt *WebsocketHandlerOption
 }
 
 type wsConfig struct {
@@ -205,7 +206,28 @@ func init() {
 	gob.Register([]any{})
 }
 
-func NewWebsocketHandler(consumer session.Consumer, redisUrl string) (*WebsocketHandler, error) {
+type WebsocketHandlerOption struct {
+	ReadBufferSize, WriteBufferSize int
+}
+
+func Option() *WebsocketHandlerOption {
+	return &WebsocketHandlerOption{
+		ReadBufferSize:  0,
+		WriteBufferSize: 0,
+	}
+}
+
+func (opt *WebsocketHandlerOption) SetReadBufferSize(size int) *WebsocketHandlerOption {
+	opt.ReadBufferSize = size
+	return opt
+}
+
+func (opt *WebsocketHandlerOption) SetWriteBufferSize(size int) *WebsocketHandlerOption {
+	opt.WriteBufferSize = size
+	return opt
+}
+
+func NewWebsocketHandler(consumer session.Consumer, redisUrl string, opts ...*WebsocketHandlerOption) (*WebsocketHandler, error) {
 	var config wsConfig
 	if err := gocommon.LoadConfig(&config); err != nil {
 		return nil, err
@@ -240,6 +262,14 @@ func NewWebsocketHandler(consumer session.Consumer, redisUrl string) (*Websocket
 		}
 	}()
 
+	var opt *WebsocketHandlerOption
+	if len(opts) > 0 {
+		// TODO : opts merge
+		opt = opts[0]
+	} else {
+		opt = Option()
+	}
+
 	ws := &WebsocketHandler{
 		websocketHandlerBase: websocketHandlerBase{
 			redisMsgChanName:  fmt.Sprintf("_wsh_msg_%d", redisSync.Options().DB),
@@ -252,6 +282,7 @@ func NewWebsocketHandler(consumer session.Consumer, redisUrl string) (*Websocket
 			sendMsgChan:       sendchan,
 			sessionConsumer:   consumer,
 		},
+		opt: opt,
 	}
 	consumer.RegisterOnSessionInvalidated(ws.onSessionInvalidated)
 	return ws, nil
@@ -266,7 +297,7 @@ func (ws *WebsocketHandler) Cleanup() {
 	ws.connWaitGroup.Wait()
 }
 
-func (ws *WebsocketHandler) RegisterHandlers(serveMux *http.ServeMux, prefix string) error {
+func (ws *WebsocketHandler) RegisterHandlers(serveMux gocommon.ServerMuxInterface, prefix string) error {
 	url := gocommon.MakeHttpHandlerPattern(prefix, "ws")
 	if *noAuthFlag {
 		serveMux.HandleFunc(url, ws.upgrade_nosession)
@@ -303,8 +334,8 @@ func (ws *WebsocketHandler) LeaveRoom(room string, accid primitive.ObjectID) {
 	}
 }
 
-func (ws *WebsocketHandler) onSessionInvalidated(accid primitive.ObjectID) {
-	ws.forceCloseChan <- accid
+func (ws *WebsocketHandler) onSessionInvalidated(invsess session.InvalidatedSession) {
+	ws.forceCloseChan <- invsess.Account
 }
 
 func (ws *WebsocketHandler) mainLoop(ctx context.Context) {
@@ -333,14 +364,15 @@ func (ws *WebsocketHandler) mainLoop(ctx context.Context) {
 				buffer := bytes.NewBuffer([]byte(raw.Payload))
 				dec := gob.NewDecoder(buffer)
 
-				if raw.Channel == ws.redisMsgChanName {
+				switch raw.Channel {
+				case ws.redisMsgChanName:
 					var msg UpstreamMessage
 					if err := dec.Decode(&msg); err == nil {
 						ws.deliveryChan <- &msg
 					} else {
 						logger.Println("decode UpstreamMessage failed :", err)
 					}
-				} else if raw.Channel == ws.redisCmdChanName {
+				case ws.redisCmdChanName:
 					var cmd commandMessage
 					if err := dec.Decode(&cmd); err == nil {
 						ws.deliveryChan <- &cmd
@@ -556,6 +588,8 @@ func (ws *WebsocketHandler) mainLoop(ctx context.Context) {
 			if c.Conn == nil {
 				delete(entireConns, c.sender.Accid.Hex())
 				go ws.ClientDisconnected(c)
+			} else if ws.sessionConsumer.IsRevoked(c.sender.Accid) {
+				c.Conn.MakeWriter().WriteControl(websocket.CloseMessage, unauthdata, time.Time{})
 			} else {
 				entireConns[c.sender.Accid.Hex()] = c
 				go ws.ClientConnected(c)
@@ -649,9 +683,9 @@ func (ws *WebsocketHandler) upgrade_nosession(w http.ResponseWriter, r *http.Req
 	accid := primitive.ObjectID(*raw)
 
 	sk := r.Header.Get("AS-X-SESSION")
-	authinfo, err := ws.sessionConsumer.Query(sk)
-	if err != nil {
-		w.WriteHeader(http.StatusBadRequest)
+	authinfo := ws.sessionConsumer.Query(sk)
+	if !authinfo.Valid() {
+		w.WriteHeader(http.StatusUnauthorized)
 		return
 	}
 
@@ -660,12 +694,10 @@ func (ws *WebsocketHandler) upgrade_nosession(w http.ResponseWriter, r *http.Req
 		return
 	}
 
-	if authinfo.Invalidated() {
-		w.WriteHeader(http.StatusUnauthorized)
-		return
-	}
-
-	var upgrader = websocket.Upgrader{} // use default options
+	var upgrader = websocket.Upgrader{
+		ReadBufferSize:  ws.opt.ReadBufferSize,
+		WriteBufferSize: ws.opt.WriteBufferSize,
+	} // use default options
 	conn, err := upgrader.Upgrade(w, r, nil)
 	if err != nil {
 		w.WriteHeader(http.StatusInternalServerError)
@@ -695,24 +727,21 @@ func (ws *WebsocketHandler) upgrade(w http.ResponseWriter, r *http.Request) {
 	}()
 
 	sk := r.Header.Get("AS-X-SESSION")
-	authinfo, err := ws.sessionConsumer.Query(sk)
-	if err != nil {
-		w.WriteHeader(http.StatusInternalServerError)
-		logger.Error("authorize query failed :", err)
-		return
-	}
-
-	if authinfo.Account.IsZero() {
+	if len(sk) == 0 {
 		w.WriteHeader(http.StatusUnauthorized)
 		return
 	}
 
-	if authinfo.Invalidated() {
+	authinfo := ws.sessionConsumer.Query(sk)
+	if !authinfo.Valid() {
 		w.WriteHeader(http.StatusUnauthorized)
 		return
 	}
 
-	var upgrader = websocket.Upgrader{} // use default options
+	var upgrader = websocket.Upgrader{
+		ReadBufferSize:  ws.opt.ReadBufferSize,
+		WriteBufferSize: ws.opt.WriteBufferSize,
+	} // use default options
 	conn, err := upgrader.Upgrade(w, r, nil)
 	if err != nil {
 		w.WriteHeader(http.StatusInternalServerError)

wshandler/wshandler_peer.go

@@ -11,6 +11,7 @@ import (
 	"time"
 
 	"go.mongodb.org/mongo-driver/bson/primitive"
+	"repositories.action2quare.com/ayo/gocommon"
 	"repositories.action2quare.com/ayo/gocommon/logger"
 	"repositories.action2quare.com/ayo/gocommon/session"
 
@@ -18,7 +19,7 @@ import (
 )
 
 type WebsocketPeerHandler interface {
-	RegisterHandlers(serveMux *http.ServeMux, prefix string) error
+	RegisterHandlers(serveMux gocommon.ServerMuxInterface, prefix string) error
 }
 
 type peerCtorChannelValue struct {
@@ -164,7 +165,7 @@ func NewWebsocketPeerHandler[T PeerInterface](consumer session.Consumer, creator
 	return wsh
 }
 
-func (ws *websocketPeerHandler[T]) RegisterHandlers(serveMux *http.ServeMux, prefix string) error {
+func (ws *websocketPeerHandler[T]) RegisterHandlers(serveMux gocommon.ServerMuxInterface, prefix string) error {
 	if *noAuthFlag {
 		serveMux.HandleFunc(prefix, ws.upgrade_noauth)
 	} else {
@@ -175,9 +176,9 @@ func (ws *websocketPeerHandler[T]) RegisterHandlers(serveMux *http.ServeMux, pre
 	return nil
 }
 
-func (ws *websocketPeerHandler[T]) onSessionInvalidated(accid primitive.ObjectID) {
+func (ws *websocketPeerHandler[T]) onSessionInvalidated(invsess session.InvalidatedSession) {
 	ws.peerDtorChannel <- peerDtorChannelValue{
-		accid: accid,
+		accid: invsess.Account,
 	}
 }
 
@@ -304,10 +305,12 @@ func (ws *websocketPeerHandler[T]) upgrade_noauth(w http.ResponseWriter, r *http
 	sk := r.Header.Get("AS-X-SESSION")
 	var accid primitive.ObjectID
 	if len(sk) > 0 {
-		authinfo, err := ws.sessionConsumer.Query(sk)
-		if err == nil {
-			accid = authinfo.Account
+		authinfo := ws.sessionConsumer.Query(sk)
+		if !authinfo.Valid() {
+			w.WriteHeader(http.StatusUnauthorized)
+			return
 		}
+		accid = authinfo.Account
 	}
 
 	if accid.IsZero() {
@@ -362,14 +365,8 @@ func (ws *websocketPeerHandler[T]) upgrade(w http.ResponseWriter, r *http.Reques
 	}()
 
 	sk := r.Header.Get("AS-X-SESSION")
-	authinfo, err := ws.sessionConsumer.Query(sk)
-	if err != nil {
-		w.WriteHeader(http.StatusInternalServerError)
-		logger.Error("authorize query failed :", err)
-		return
-	}
-
-	if authinfo.Account.IsZero() || authinfo.Invalidated() {
+	authinfo := ws.sessionConsumer.Query(sk)
+	if !authinfo.Valid() {
 		w.WriteHeader(http.StatusUnauthorized)
 		return
 	}