Merge branch 'master' of https://repositories.action2quare.com/ayo/gocommon

- Email -> Alias

logger/logger.go

@@ -8,6 +8,7 @@ import (
 	"path"
 	"runtime"
 	"runtime/debug"
+	"strconv"
 	"strings"
 
 	"repositories.action2quare.com/ayo/gocommon/flagx"
@@ -15,6 +16,7 @@ import (
 
 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()
@@ -23,7 +25,29 @@ func init() {
 	var outWriter io.Writer
 	outWriter = os.Stdout
 
-	if *UseLogFile {
+	args := os.Args
+	useLogFile := false
+	for _, arg := range args {
+		if strings.HasPrefix(arg, "-logfile=") {
+			useLogFile, _ = strconv.ParseBool(arg[9:])
+			break
+		}
+
+		if arg == "-logfile" {
+			useLogFile = true
+			break
+		}
+	}
+
+	logprefix := 3
+	for _, arg := range args {
+		if strings.HasPrefix(arg, "-logprefix=") {
+			logprefix, _ = strconv.Atoi(arg[11:])
+			break
+		}
+	}
+
+	if useLogFile {
 		ext := path.Ext(binname)
 		if len(ext) > 0 {
 			binname = binname[:len(binname)-len(ext)]
@@ -38,7 +62,11 @@ func init() {
 		outWriter = io.MultiWriter(outWriter, logFile)
 	}
 
+	if logprefix < 4 {
+		stdlogger = log.New(outWriter, "", logprefix)
+	} else {
 		stdlogger = log.New(outWriter, "", log.LstdFlags)
+	}
 }
 
 func Println(v ...interface{}) {

metric/prometheus.go

@@ -135,7 +135,11 @@ func (pe *prometheusExporter) loop(ctx context.Context) {
 			}
 
 			if err := prometheus.Register(nextcollector); err != nil {
+				if _, ok := err.(prometheus.AlreadyRegisteredError); ok {
+					// 이미 등록된 metric. child process를 여럿 실행하면 발생됨
+				} else {
 					logger.Error("prometheus register err :", *nm, err)
+				}
 			} else {
 				collector = nextcollector
 			}

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)}
 	}
+
+	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 {

server.go

@@ -40,6 +40,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 +146,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 +166,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 {
@@ -481,6 +504,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

@@ -19,7 +19,7 @@ type Authorization struct {
 	// by authorization provider
 	Platform string `bson:"p" json:"p"`
 	Uid      string `bson:"u" json:"u"`
-	Email    string `bson:"em" json:"em"`
+	Alias    string `bson:"al" json:"al"`
 }
 
 func (auth *Authorization) ToStrings() []string {
@@ -27,7 +27,7 @@ func (auth *Authorization) ToStrings() []string {
 		"a", auth.Account.Hex(),
 		"p", auth.Platform,
 		"u", auth.Uid,
-		"em", auth.Email,
+		"al", auth.Alias,
 		"inv", auth.invalidated,
 	}
 }
@@ -42,7 +42,7 @@ func MakeAuthrizationFromStringMap(src map[string]string) Authorization {
 		Account:     accid,
 		Platform:    src["p"],
 		Uid:         src["u"],
-		Email:       src["em"],
+		Alias:       src["al"],
 		invalidated: src["inv"],
 	}
 }

wshandler/wshandler.go

@@ -266,7 +266,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)
@@ -586,9 +586,11 @@ func upgrade_core(ws *WebsocketHandler, conn *Conn, accid primitive.ObjectID, al
 			ws.connWaitGroup.Done()
 		}()
 
+		logger.Println("wshandler connected :", accid)
 		for {
 			messageType, r, err := c.innerConn.NextReader()
 			if err != nil {
+				logger.Println("wshandler NextReader err :", messageType, err)
 				if ce, ok := err.(*websocket.CloseError); ok {
 					c.closeMessage = ce.Text
 				}
@@ -610,6 +612,7 @@ func upgrade_core(ws *WebsocketHandler, conn *Conn, accid primitive.ObjectID, al
 				ws.Call(newconn.sender, string(cmd), r)
 			}
 		}
+		logger.Println("wshandler disconnected :", accid, c.closeMessage)
 		c.Conn = nil
 		if c.closeMessage != ForceShutdownCloseMessage {
 			ws.connInOutChan <- c

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 {