<commit_before>package rest

import (
	"bytes"
	"encoding/json"
	"fmt"
	"net/http"
	"regexp"
	"runtime/debug"
	"strconv"
	"strings"
	"sync"
	"sync/atomic"
	"time"

	"github.com/couchbase/go-blip"
	"github.com/couchbase/sync_gateway/auth"
	"github.com/couchbase/sync_gateway/base"
	"github.com/couchbase/sync_gateway/channels"
	"github.com/couchbase/sync_gateway/db"
	"golang.org/x/net/websocket"
)

const (

	// Blip default vals
	BlipDefaultBatchSize = uint64(200)

	BlipMinimumBatchSize = uint64(10) // Not in the replication spec - is this required?

	// The AppProtocolId part of the BLIP websocket subprotocol.  Must match identically with the peer (typically CBLite / LiteCore).
	// At some point this will need to be able to support an array of protocols.  See go-blip/issues/27.
	BlipCBMobileReplication = "CBMobile_2"
)

// Represents one BLIP connection (socket) opened by a client.
// This connection remains open until the client closes it, and can receive any number of requests.
type blipSyncContext struct {
	blipContext         *blip.Context
	dbc                 *db.DatabaseContext
	user                auth.User
	effectiveUsername   string
	batchSize           int
	continuous          bool
	activeOnly          bool
	channels            base.Set
	lock                sync.Mutex
	allowedAttachments  map[string]int
	handlerSerialNumber uint64    // Each handler within a context gets a unique serial number for logging
	terminator          chan bool // Closed during blipSyncContext.close(). Ensures termination of async goroutines.
}

type blipHandler struct {
	*blipSyncContext
	db           *db.Database
	serialNumber uint64 // This blip handler's serial number to differentiate logs w/ other handlers
}

type blipHandlerMethod func(*blipHandler, *blip.Message) error

// Wrap blip handler with code that reloads the user object to make sure that
// it has the latest channel access grants.
func userBlipHandler(underlyingMethod blipHandlerMethod) blipHandlerMethod {

	wrappedBlipHandler := func(bh *blipHandler, bm *blip.Message) error {

		// Reload the user on each blip request (otherwise runs into SG issue #2717)
		if err := bh.db.ReloadUser(); err != nil {
			return err
		}

		// Call down to underlying method and return it's value
		return underlyingMethod(bh, bm)
	}

	return wrappedBlipHandler

}

// Maps the profile (verb) of an incoming request to the method that handles it.
var kHandlersByProfile = map[string]blipHandlerMethod{
	"getCheckpoint":  (*blipHandler).handleGetCheckpoint,
	"setCheckpoint":  (*blipHandler).handleSetCheckpoint,
	"subChanges":     userBlipHandler((*blipHandler).handleSubChanges),
	"changes":        userBlipHandler((*blipHandler).handleChanges),
	"rev":            userBlipHandler((*blipHandler).handleRev),
	"getAttachment":  userBlipHandler((*blipHandler).handleGetAttachment),
	"proposeChanges": (*blipHandler).handleProposedChanges,
}

// HTTP handler for incoming BLIP sync WebSocket request (/db/_blipsync)
func (h *handler) handleBLIPSync() error {
	if c := h.server.GetConfig().ReplicatorCompression; c != nil {
		blip.CompressionLevel = *c
	}

	// Create a BLIP context:
	blipContext := blip.NewContext(BlipCBMobileReplication)
	blipContext.Logger = DefaultBlipLogger(blipContext.ID)
	blipContext.LogMessages = base.LogEnabledExcludingLogStar("WS+")
	blipContext.LogFrames = base.LogEnabledExcludingLogStar("WS++")

	// Create a BLIP-sync context and register handlers:
	ctx := blipSyncContext{
		blipContext:       blipContext,
		dbc:               h.db.DatabaseContext,
		user:              h.user,
		effectiveUsername: h.currentEffectiveUserName(),
		terminator:        make(chan bool),
	}
	defer ctx.close()

	blipContext.DefaultHandler = ctx.notFound
	for profile, handlerFn := range kHandlersByProfile {
		ctx.register(profile, handlerFn)
	}

	ctx.blipContext.FatalErrorHandler = func(err error) {
		ctx.LogTo("HTTP", "#%03d:     --> BLIP+WebSocket connection error: %v", h.serialNumber, err)
	}

	// Create a BLIP WebSocket handler and have it handle the request:
	server := blipContext.WebSocketServer()
	defaultHandler := server.Handler
	server.Handler = func(conn *websocket.Conn) {
		h.logStatus(101, fmt.Sprintf("[%s] Upgraded to BLIP+WebSocket protocol. User:%s.", blipContext.ID, h.currentEffectiveUserName()))
		defer func() {
			conn.Close() // in case it wasn't closed already
			ctx.LogTo("HTTP+", "#%03d:    --> BLIP+WebSocket connection closed", h.serialNumber)
		}()
		defaultHandler(conn)
	}

	server.ServeHTTP(h.response, h.rq)
	return nil
}

// Registers a BLIP handler including the outer-level work of logging & error handling.
// Includes the outer handler as a nested function.
func (ctx *blipSyncContext) register(profile string, handlerFn func(*blipHandler, *blip.Message) error) {

	// Wrap the handler function with a function that adds handling needed by all handlers
	handlerFnWrapper := func(rq *blip.Message) {

		startTime := time.Now()

		db, _ := db.GetDatabase(ctx.dbc, ctx.user)
		handler := blipHandler{
			blipSyncContext: ctx,
			db:              db,
			serialNumber:    ctx.incrementSerialNumber(),
		}

		if err := handlerFn(&handler, rq); err != nil {
			status, msg := base.ErrorAsHTTPStatus(err)
			if response := rq.Response(); response != nil {
				response.SetError("HTTP", status, msg)
			}
			ctx.LogTo("SyncMsg", "#%03d: Type:%s   --> %d %s Time:%v User:%s", handler.serialNumber, profile, status, msg, time.Since(startTime), ctx.effectiveUsername)
		} else {

			// Log the fact that the handler has finished, except for the "subChanges" special case which does it's own termination related logging
			if profile != "subChanges" {
				ctx.LogTo("SyncMsg+", "#%03d: Type:%s   --> OK Time:%v User:%s ", handler.serialNumber, profile, time.Since(startTime), ctx.effectiveUsername)
			}
		}
	}

	ctx.blipContext.HandlerForProfile[profile] = handlerFnWrapper

}

func (ctx *blipSyncContext) close() {
	close(ctx.terminator)
}

// Handler for unknown requests
func (ctx *blipSyncContext) notFound(rq *blip.Message) {
	ctx.LogTo("Sync", "%s Type:%q User:%s", rq, rq.Profile(), ctx.effectiveUsername)
	ctx.LogTo("Sync", "%s    --> 404 Unknown profile. User:%s", rq, ctx.effectiveUsername)
	blip.Unhandled(rq)
}

func (ctx *blipSyncContext) LogTo(key string, format string, args ...interface{}) {
	formatWithContextID, paramsWithContextID := base.PrependContextID(ctx.blipContext.ID, format, args...)
	base.LogTo(key, formatWithContextID, paramsWithContextID...)
}

//////// CHECKPOINTS

// Received a "getCheckpoint" request
func (bh *blipHandler) handleGetCheckpoint(rq *blip.Message) error {

	client := rq.Properties["client"]
	bh.logEndpointEntry(rq.Profile(), fmt.Sprintf("Client:%s", client))

	docID := fmt.Sprintf("checkpoint/%s", client)
	response := rq.Response()
	if response == nil {
		return nil
	}

	value, err := bh.db.GetSpecial("local", docID)
	if err != nil {
		return err
	}
	if value == nil {
		return base.HTTPErrorf(http.StatusNotFound, http.StatusText(http.StatusNotFound))
	}
	response.Properties["rev"] = value["_rev"].(string)
	delete(value, "_rev")
	delete(value, "_id")
	value.FixJSONNumbers()
	response.SetJSONBody(value)
	return nil
}

// Received a "setCheckpoint" request
func (bh *blipHandler) handleSetCheckpoint(rq *blip.Message) error {

	setCheckpointParams := newSetCheckpointParams(rq)
	bh.logEndpointEntry(rq.Profile(), setCheckpointParams.String())

	docID := fmt.Sprintf("checkpoint/%s", setCheckpointParams.client())

	var checkpoint db.Body
	if err := rq.ReadJSONBody(&checkpoint); err != nil {
		return err
	}
	if revID := setCheckpointParams.rev(); revID != "" {
		checkpoint["_rev"] = revID
	}
	revID, err := bh.db.PutSpecial("local", docID, checkpoint)
	if err != nil {
		return err
	}
	rq.Response().Properties["rev"] = revID
	return nil
}

//////// CHANGES

// Received a "subChanges" subscription request
func (bh *blipHandler) handleSubChanges(rq *blip.Message) error {

	subChangesParams, err := newSubChangesParams(
		rq,
		bh.blipSyncContext,
		bh.db.CreateZeroSinceValue(),
		bh.db.ParseSequenceID,
	)
	if err != nil {
		return base.HTTPErrorf(http.StatusBadRequest, "Invalid subChanges parameters")
	}

	if len(subChangesParams.docIDs()) > 0 && subChangesParams.continuous() {
		return base.HTTPErrorf(http.StatusBadRequest, "DocIDs filter not supported for continuous subChanges")
	}

	bh.logEndpointEntry(rq.Profile(), subChangesParams.String())

	// TODO: Do we need to store the changes-specific parameters on the blip sync context?  Seems like they only need to be passed in to sendChanges
	bh.batchSize = subChangesParams.batchSize()
	bh.continuous = subChangesParams.continuous()
	bh.activeOnly = subChangesParams.activeOnly()

	if filter := subChangesParams.filter(); filter == "sync_gateway/bychannel" {
		var err error

		bh.channels, err = subChangesParams.channelsExpandedSet()
		if err != nil {
			return base.HTTPErrorf(http.StatusBadRequest, "%s", err)
		} else if len(bh.channels) == 0 {
			return base.HTTPErrorf(http.StatusBadRequest, "Empty channel list")

		}
	} else if filter != "" {
		return base.HTTPErrorf(http.StatusBadRequest, "Unknown filter; try sync_gateway/bychannel")
	}

	// Start asynchronous changes goroutine
	go func() {
		base.StatsExpvars.Add("subChanges_total", 1)
		base.StatsExpvars.Add("subChanges_active", 1)
		defer base.StatsExpvars.Add("subChanges_active", -1)
		// sendChanges runs until blip context closes, or fails due to error
		startTime := time.Now()
		bh.sendChanges(rq.Sender, subChangesParams)
		bh.LogTo("SyncMsg+", "#%03d: Type:%s   --> Time:%v User:%s ", bh.serialNumber, rq.Profile(), time.Since(startTime), bh.effectiveUsername)
	}()

	return nil
}

// Sends all changes since the given sequence
func (bh *blipHandler) sendChanges(sender *blip.Sender, params *subChangesParams) {
	defer func() {
		if panicked := recover(); panicked != nil {
			base.Warn("[%s] PANIC sending changes: %v\n%s", bh.blipContext.ID, panicked, debug.Stack())
		}
	}()

	bh.LogTo("Sync", "Sending changes since %v. User:%s", params.since(), bh.effectiveUsername)
	options := db.ChangesOptions{
		Since:      params.since(),
		Conflicts:  true,
		Continuous: bh.continuous,
		ActiveOnly: bh.activeOnly,
		Terminator: bh.blipSyncContext.terminator,
	}

	channelSet := bh.channels
	if channelSet == nil {
		channelSet = channels.SetOf(channels.AllChannelWildcard)
	}
<commit_msg>Improves situation on LiteCore #437 (#3349)<commit_after>package rest

import (
	"bytes"
	"encoding/json"
	"fmt"
	"net/http"
	"regexp"
	"runtime/debug"
	"strconv"
	"strings"
	"sync"
	"sync/atomic"
	"time"

	"github.com/couchbase/go-blip"
	"github.com/couchbase/sync_gateway/auth"
	"github.com/couchbase/sync_gateway/base"
	"github.com/couchbase/sync_gateway/channels"
	"github.com/couchbase/sync_gateway/db"
	"golang.org/x/net/websocket"
)

const (

	// Blip default vals
	BlipDefaultBatchSize = uint64(200)

	BlipMinimumBatchSize = uint64(10) // Not in the replication spec - is this required?

	// The AppProtocolId part of the BLIP websocket subprotocol.  Must match identically with the peer (typically CBLite / LiteCore).
	// At some point this will need to be able to support an array of protocols.  See go-blip/issues/27.
	BlipCBMobileReplication = "CBMobile_2"
)

// Represents one BLIP connection (socket) opened by a client.
// This connection remains open until the client closes it, and can receive any number of requests.
type blipSyncContext struct {
	blipContext         *blip.Context
	dbc                 *db.DatabaseContext
	user                auth.User
	effectiveUsername   string
	batchSize           int
	continuous          bool
	activeOnly          bool
	channels            base.Set
	lock                sync.Mutex
	allowedAttachments  map[string]int
	handlerSerialNumber uint64    // Each handler within a context gets a unique serial number for logging
	terminator          chan bool // Closed during blipSyncContext.close(). Ensures termination of async goroutines.
}

type blipHandler struct {
	*blipSyncContext
	db           *db.Database
	serialNumber uint64 // This blip handler's serial number to differentiate logs w/ other handlers
}

type blipHandlerMethod func(*blipHandler, *blip.Message) error

// Wrap blip handler with code that reloads the user object to make sure that
// it has the latest channel access grants.
func userBlipHandler(underlyingMethod blipHandlerMethod) blipHandlerMethod {

	wrappedBlipHandler := func(bh *blipHandler, bm *blip.Message) error {

		// Reload the user on each blip request (otherwise runs into SG issue #2717)
		if err := bh.db.ReloadUser(); err != nil {
			return err
		}

		// Call down to underlying method and return it's value
		return underlyingMethod(bh, bm)
	}

	return wrappedBlipHandler

}

// Maps the profile (verb) of an incoming request to the method that handles it.
var kHandlersByProfile = map[string]blipHandlerMethod{
	"getCheckpoint":  (*blipHandler).handleGetCheckpoint,
	"setCheckpoint":  (*blipHandler).handleSetCheckpoint,
	"subChanges":     userBlipHandler((*blipHandler).handleSubChanges),
	"changes":        userBlipHandler((*blipHandler).handleChanges),
	"rev":            userBlipHandler((*blipHandler).handleRev),
	"getAttachment":  userBlipHandler((*blipHandler).handleGetAttachment),
	"proposeChanges": (*blipHandler).handleProposedChanges,
}

// HTTP handler for incoming BLIP sync WebSocket request (/db/_blipsync)
func (h *handler) handleBLIPSync() error {
	if c := h.server.GetConfig().ReplicatorCompression; c != nil {
		blip.CompressionLevel = *c
	}

	// Create a BLIP context:
	blipContext := blip.NewContext(BlipCBMobileReplication)
	blipContext.Logger = DefaultBlipLogger(blipContext.ID)
	blipContext.LogMessages = base.LogEnabledExcludingLogStar("WS+")
	blipContext.LogFrames = base.LogEnabledExcludingLogStar("WS++")

	// Create a BLIP-sync context and register handlers:
	ctx := blipSyncContext{
		blipContext:       blipContext,
		dbc:               h.db.DatabaseContext,
		user:              h.user,
		effectiveUsername: h.currentEffectiveUserName(),
		terminator:        make(chan bool),
	}
	defer ctx.close()

	blipContext.DefaultHandler = ctx.notFound
	for profile, handlerFn := range kHandlersByProfile {
		ctx.register(profile, handlerFn)
	}

	ctx.blipContext.FatalErrorHandler = func(err error) {
		ctx.LogTo("HTTP", "#%03d:     --> BLIP+WebSocket connection error: %v", h.serialNumber, err)
	}

	// Create a BLIP WebSocket handler and have it handle the request:
	server := blipContext.WebSocketServer()
	defaultHandler := server.Handler
	server.Handler = func(conn *websocket.Conn) {
		h.logStatus(101, fmt.Sprintf("[%s] Upgraded to BLIP+WebSocket protocol. User:%s.", blipContext.ID, h.currentEffectiveUserName()))
		defer func() {
			conn.Close() // in case it wasn't closed already
			ctx.LogTo("HTTP+", "#%03d:    --> BLIP+WebSocket connection closed", h.serialNumber)
		}()
		defaultHandler(conn)
	}

	server.ServeHTTP(h.response, h.rq)
	return nil
}

// Registers a BLIP handler including the outer-level work of logging & error handling.
// Includes the outer handler as a nested function.
func (ctx *blipSyncContext) register(profile string, handlerFn func(*blipHandler, *blip.Message) error) {

	// Wrap the handler function with a function that adds handling needed by all handlers
	handlerFnWrapper := func(rq *blip.Message) {

		startTime := time.Now()

		db, _ := db.GetDatabase(ctx.dbc, ctx.user)
		handler := blipHandler{
			blipSyncContext: ctx,
			db:              db,
			serialNumber:    ctx.incrementSerialNumber(),
		}

		if err := handlerFn(&handler, rq); err != nil {
			status, msg := base.ErrorAsHTTPStatus(err)
			if response := rq.Response(); response != nil {
				response.SetError("HTTP", status, msg)
			}
			ctx.LogTo("SyncMsg", "#%03d: Type:%s   --> %d %s Time:%v User:%s", handler.serialNumber, profile, status, msg, time.Since(startTime), ctx.effectiveUsername)
		} else {

			// Log the fact that the handler has finished, except for the "subChanges" special case which does it's own termination related logging
			if profile != "subChanges" {
				ctx.LogTo("SyncMsg+", "#%03d: Type:%s   --> OK Time:%v User:%s ", handler.serialNumber, profile, time.Since(startTime), ctx.effectiveUsername)
			}
		}
	}

	ctx.blipContext.HandlerForProfile[profile] = handlerFnWrapper

}

func (ctx *blipSyncContext) close() {
	close(ctx.terminator)
}

// Handler for unknown requests
func (ctx *blipSyncContext) notFound(rq *blip.Message) {
	ctx.LogTo("Sync", "%s Type:%q User:%s", rq, rq.Profile(), ctx.effectiveUsername)
	ctx.LogTo("Sync", "%s    --> 404 Unknown profile. User:%s", rq, ctx.effectiveUsername)
	blip.Unhandled(rq)
}

func (ctx *blipSyncContext) LogTo(key string, format string, args ...interface{}) {
	formatWithContextID, paramsWithContextID := base.PrependContextID(ctx.blipContext.ID, format, args...)
	base.LogTo(key, formatWithContextID, paramsWithContextID...)
}

//////// CHECKPOINTS

// Received a "getCheckpoint" request
func (bh *blipHandler) handleGetCheckpoint(rq *blip.Message) error {

	client := rq.Properties["client"]
	bh.logEndpointEntry(rq.Profile(), fmt.Sprintf("Client:%s", client))

	docID := fmt.Sprintf("checkpoint/%s", client)
	response := rq.Response()
	if response == nil {
		return nil
	}

	value, err := bh.db.GetSpecial("local", docID)
	if err != nil {
		return err
	}
	if value == nil {
		return base.HTTPErrorf(http.StatusNotFound, http.StatusText(http.StatusNotFound))
	}
	response.Properties["rev"] = value["_rev"].(string)
	delete(value, "_rev")
	delete(value, "_id")
	value.FixJSONNumbers()
	response.SetJSONBody(value)
	return nil
}

// Received a "setCheckpoint" request
func (bh *blipHandler) handleSetCheckpoint(rq *blip.Message) error {

	setCheckpointParams := newSetCheckpointParams(rq)
	bh.logEndpointEntry(rq.Profile(), setCheckpointParams.String())

	docID := fmt.Sprintf("checkpoint/%s", setCheckpointParams.client())

	var checkpoint db.Body
	if err := rq.ReadJSONBody(&checkpoint); err != nil {
		return err
	}
	if revID := setCheckpointParams.rev(); revID != "" {
		checkpoint["_rev"] = revID
	}
	revID, err := bh.db.PutSpecial("local", docID, checkpoint)
	if err != nil {
		return err
	}
	rq.Response().Properties["rev"] = revID
	return nil
}

//////// CHANGES

// Received a "subChanges" subscription request
func (bh *blipHandler) handleSubChanges(rq *blip.Message) error {

	subChangesParams, err := newSubChangesParams(
		rq,
		bh.blipSyncContext,
		bh.db.CreateZeroSinceValue(),
		bh.db.ParseSequenceID,
	)
	if err != nil {
		return base.HTTPErrorf(http.StatusBadRequest, "Invalid subChanges parameters")
	}

	if len(subChangesParams.docIDs()) > 0 && subChangesParams.continuous() {
		return base.HTTPErrorf(http.StatusBadRequest, "DocIDs filter not supported for continuous subChanges")
	}

	bh.logEndpointEntry(rq.Profile(), subChangesParams.String())

	// TODO: Do we need to store the changes-specific parameters on the blip sync context?  Seems like they only need to be passed in to sendChanges
	bh.batchSize = subChangesParams.batchSize()
	bh.continuous = subChangesParams.continuous()
	bh.activeOnly = subChangesParams.activeOnly()

	if filter := subChangesParams.filter(); filter == "sync_gateway/bychannel" {
		var err error

		bh.channels, err = subChangesParams.channelsExpandedSet()
		if err != nil {
			return base.HTTPErrorf(http.StatusBadRequest, "%s", err)
		} else if len(bh.channels) == 0 {
			return base.HTTPErrorf(http.StatusBadRequest, "Empty channel list")

		}
	} else if filter != "" {
		return base.HTTPErrorf(http.StatusBadRequest, "Unknown filter; try sync_gateway/bychannel")
	}

	// Start asynchronous changes goroutine
	go func() {
		base.StatsExpvars.Add("subChanges_total", 1)
		base.StatsExpvars.Add("subChanges_active", 1)
		defer base.StatsExpvars.Add("subChanges_active", -1)
		// sendChanges runs until blip context closes, or fails due to error
		startTime := time.Now()
		bh.sendChanges(rq.Sender, subChangesParams)
		bh.LogTo("SyncMsg+", "#%03d: Type:%s   --> Time:%v User:%s ", bh.serialNumber, rq.Profile(), time.Since(startTime), bh.effectiveUsername)
	}()

	return nil
}

// Sends all changes since the given sequence
func (bh *blipHandler) sendChanges(sender *blip.Sender, params *subChangesParams) {
	defer func() {
		if panicked := recover(); panicked != nil {
			base.Warn("[%s] PANIC sending changes: %v\n%s", bh.blipContext.ID, panicked, debug.Stack())
		}
	}()

	// Don't send conflicting rev tree branches, just send the winning revision + history, since
	// CBL 2.0 (and blip_sync) don't support branched revision trees.  See LiteCore #437.
	sendConflicts := false

	bh.LogTo("Sync", "Sending changes since %v. User:%s", params.since(), bh.effectiveUsername)
	options := db.ChangesOptions{
		Since:      params.since(),
		Conflicts:  sendConflicts,
		Continuous: bh.continuous,
		ActiveOnly: bh.activeOnly,
		Terminator: bh.blipSyncContext.terminator,
	}

	channelSet := bh.channels
	if channelSet == nil {
		channelSet = channels.SetOf(channels.AllChannelWildcard)
	}

{'fraction_non_alphanumeric': 0.12268859628273994, 'fraction_numerical': 0.004468317725911489, 'mean_word_length': 4.208715028472394, 'pattern_counts': {'":': 14, '<': 3, '<?xml version=': 0, '>': 17, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1729811', 'n_tokens_mistral': 7063, 'n_tokens_neox': 6691, 'n_words': 1756}