// Top-level method to get all the changes in a channel since the sequence 'since'.
// If the cache doesn't go back far enough, the view will be queried.
// View query results may be fed back into the cache if there's room.
// initialSequence is used only if the cache is empty: it gives the max sequence to which the
// view should be queried, because we don't want the view query to outrun the chanceCache's
// nextSequence.
func (c *channelCache) GetChanges(options ChangesOptions) ([]*LogEntry, error) {
	// Use the cache, and return if it fulfilled the entire request:
	cacheValidFrom, resultFromCache := c.getCachedChanges(options)
	numFromCache := len(resultFromCache)
	if numFromCache > 0 || resultFromCache == nil {
		base.LogTo("Cache", "getCachedChanges(%q, %d) --> %d changes valid from #%d",
			c.channelName, options.Since, numFromCache, cacheValidFrom)
	} else if resultFromCache == nil {
		base.LogTo("Cache", "getCachedChanges(%q, %d) --> nothing cached",
			c.channelName, options.Since)
	}
	if cacheValidFrom <= options.Since+1 {
		return resultFromCache, nil
	}

	// Nope, we're going to have to backfill from the view.
	//** First acquire the _view_ lock (not the regular lock!)
	c.viewLock.Lock()
	defer c.viewLock.Unlock()

	// Another goroutine might have gotten the lock first and already queried the view and updated
	// the cache, so repeat the above:
	cacheValidFrom, resultFromCache = c._getCachedChanges(options)
	if len(resultFromCache) > numFromCache {
		base.LogTo("Cache", "2nd getCachedChanges(%q, %d) got %d more, valid from #%d!",
			c.channelName, options.Since, len(resultFromCache)-numFromCache, cacheValidFrom)
	}
	if cacheValidFrom <= options.Since+1 {
		return resultFromCache, nil
	}

	// Now query the view. We set the max sequence equal to cacheValidFrom, so we'll get one
	// overlap, which helps confirm that we've got everything.
	resultFromView, err := c.context.getChangesInChannelFromView(c.channelName, cacheValidFrom,
		options)
	if err != nil {
		return nil, err
	}

	// Cache some of the view results, if there's room in the cache:
	if len(resultFromCache) < ChannelCacheMaxLength {
		c.prependChanges(resultFromView, options.Since+1, options.Limit == 0)
	}

	result := resultFromView
	room := options.Limit - len(result)
	if (options.Limit == 0 || room > 0) && len(resultFromCache) > 0 {
		// Concatenate the view & cache results:
		if len(result) > 0 && resultFromCache[0].Sequence == result[len(result)-1].Sequence {
			resultFromCache = resultFromCache[1:]
		}
		n := len(resultFromCache)
		if options.Limit > 0 && room > 0 && room < n {
			n = room
		}
		result = append(result, resultFromCache[0:n]...)
	}
	base.LogTo("Cache", "GetChangesInChannel(%q) --> %d rows", c.channelName, len(result))
	return result, nil
}

// Main loop that pulls changes from the external bucket. (Runs in its own goroutine.)
func (s *Shadower) readTapFeed() {
	vbucketsFilling := 0
	for event := range s.tapFeed.Events() {
		switch event.Opcode {
		case walrus.TapBeginBackfill:
			if vbucketsFilling == 0 {
				base.LogTo("Shadow", "Reading history of external bucket")
			}
			vbucketsFilling++
			//base.LogTo("Shadow", "Reading history of external bucket")
		case walrus.TapMutation, walrus.TapDeletion:
			key := string(event.Key)
			// Ignore ephemeral documents or ones whose ID would conflict with our metadata
			if event.Expiry > 0 || !s.docIDMatches(key) {
				break
			}
			isDeletion := event.Opcode == walrus.TapDeletion
			err := s.pullDocument(key, event.Value, isDeletion, event.Sequence, event.Flags)
			if err != nil {
				base.Warn("Error applying change from external bucket: %v", err)
			}
		case walrus.TapEndBackfill:
			if vbucketsFilling--; vbucketsFilling == 0 {
				base.LogTo("Shadow", "Caught up with history of external bucket")
			}
		}
	}
	base.LogTo("Shadow", "End of tap feed(?)")
}

// Main loop that pulls changes from the external bucket. (Runs in its own goroutine.)
func (s *Shadower) readTapFeed() {
	vbucketsFilling := 0
	for event := range s.tapFeed.Events() {
		switch event.Opcode {
		case walrus.TapBeginBackfill:
			if vbucketsFilling == 0 {
				base.LogTo("Shadow", "Reading history of external bucket")
			}
			vbucketsFilling++
			//base.LogTo("Shadow", "Reading history of external bucket")
		case walrus.TapMutation, walrus.TapDeletion:
			key := string(event.Key)
			if !s.docIDMatches(key) {
				break
			}
			isDeletion := event.Opcode == walrus.TapDeletion
			if !isDeletion && event.Expiry > 0 {
				break // ignore ephemeral documents
			}
			err := s.pullDocument(key, event.Value, isDeletion, event.Sequence, event.Flags)
			if err != nil {
				base.Warn("Error applying change from external bucket: %v", err)
			}
			atomic.AddUint64(&s.pullCount, 1)
		case walrus.TapEndBackfill:
			if vbucketsFilling--; vbucketsFilling == 0 {
				base.LogTo("Shadow", "Caught up with history of external bucket")
			}
		}
	}
	base.LogTo("Shadow", "End of tap feed(?)")
}

// Saves a new local revision to the external bucket.
func (s *Shadower) PushRevision(doc *document) {
	defer func() { atomic.AddUint64(&s.pushCount, 1) }()
	if !s.docIDMatches(doc.ID) {
		return
	} else if doc.newestRevID() == doc.UpstreamRev {
		return // This revision was pulled from the external bucket, so don't push it back!
	}

	var err error
	if doc.Flags&channels.Deleted != 0 {
		base.LogTo("Shadow", "Pushing %q, rev %q [deletion]", doc.ID, doc.CurrentRev)
		err = s.bucket.Delete(doc.ID)
	} else {
		base.LogTo("Shadow", "Pushing %q, rev %q", doc.ID, doc.CurrentRev)
		body := doc.getRevision(doc.CurrentRev)
		if body == nil {
			base.Warn("Can't get rev %q.%q to push to external bucket", doc.ID, doc.CurrentRev)
			return
		}
		err = s.bucket.Set(doc.ID, 0, body)
	}
	if err != nil {
		base.Warn("Error pushing rev of %q to external bucket: %v", doc.ID, err)
	}
}

func (db *Database) WaitForRevision() bool {
	base.LogTo("Changes", "\twaiting for a revision...")
	db.tapNotifier.L.Lock()
	defer db.tapNotifier.L.Unlock()
	db.tapNotifier.Wait()
	base.LogTo("Changes", "\t...done waiting")
	return true
}

// Given a newly changed document (received from the tap feed), adds change entries to channels.
// The JSON must be the raw document from the bucket, with the metadata and all.
func (c *changeCache) DocChanged(docID string, docJSON []byte) {
	entryTime := time.Now()
	// ** This method does not directly access any state of c, so it doesn't lock.
	go func() {
		// Is this a user/role doc?
		if strings.HasPrefix(docID, auth.UserKeyPrefix) {
			c.processPrincipalDoc(docID, docJSON, true)
			return
		} else if strings.HasPrefix(docID, auth.RoleKeyPrefix) {
			c.processPrincipalDoc(docID, docJSON, false)
			return
		}

		// First unmarshal the doc (just its metadata, to save time/memory):
		doc, err := unmarshalDocumentSyncData(docJSON, false)
		if err != nil || !doc.hasValidSyncData() {
			base.Warn("changeCache: Error unmarshaling doc %q: %v", docID, err)
			return
		}

		if doc.Sequence <= c.initialSequence {
			return // Tap is sending us an old value from before I started up; ignore it
		}

		// Record a histogram of the Tap feed's lag:
		tapLag := time.Since(doc.TimeSaved) - time.Since(entryTime)
		lagMs := int(tapLag/(100*time.Millisecond)) * 100
		changeCacheExpvars.Add(fmt.Sprintf("lag-tap-%04dms", lagMs), 1)

		// If the doc update wasted any sequences due to conflicts, add empty entries for them:
		for _, seq := range doc.UnusedSequences {
			base.LogTo("Cache", "Received unused #%d for (%q / %q)", seq, docID, doc.CurrentRev)
			change := &LogEntry{
				Sequence:     seq,
				TimeReceived: time.Now(),
			}
			c.processEntry(change)
		}

		// Now add the entry for the new doc revision:
		change := &LogEntry{
			Sequence:     doc.Sequence,
			DocID:        docID,
			RevID:        doc.CurrentRev,
			Flags:        doc.Flags,
			TimeReceived: time.Now(),
			TimeSaved:    doc.TimeSaved,
			Channels:     doc.Channels,
		}
		base.LogTo("Cache", "Received #%d after %3dms (%q / %q)", change.Sequence, int(tapLag/time.Millisecond), change.DocID, change.RevID)

		changedChannels := c.processEntry(change)
		if c.onChange != nil && len(changedChannels) > 0 {
			c.onChange(changedChannels)
		}
	}()
}

// Prepends an array of entries to this one, skipping ones that I already have.
// The new array needs to overlap with my current log, i.e. must contain the same sequence as
// c.logs[0], otherwise nothing will be added because the method can't confirm that there are no
// missing sequences in between.
// Returns the number of entries actually prepended.
func (c *channelCache) prependChanges(changes LogEntries, changesValidFrom uint64, openEnded bool) int {
	c.lock.Lock()
	defer c.lock.Unlock()

	log := c.logs
	if len(log) == 0 {
		// If my cache is empty, just copy the new changes:
		if len(changes) > 0 {
			if !openEnded && changes[len(changes)-1].Sequence < c.validFrom {
				return 0 // changes might not go all the way to the current time
			}
			if excess := len(changes) - ChannelCacheMaxLength; excess > 0 {
				changes = changes[excess:]
				changesValidFrom = changes[0].Sequence
			}
			c.logs = make(LogEntries, len(changes))
			copy(c.logs, changes)
			base.LogTo("Cache", "  Initialized cache of %q with %d entries from view (#%d--#%d)",
				c.channelName, len(changes), changes[0].Sequence, changes[len(changes)-1].Sequence)
		}
		c.validFrom = changesValidFrom
		return len(changes)

	} else if len(changes) == 0 {
		if openEnded && changesValidFrom < c.validFrom {
			c.validFrom = changesValidFrom
		}
		return 0

	} else {
		// Look for an overlap, and prepend everything up to that point:
		firstSequence := log[0].Sequence
		if changes[0].Sequence <= firstSequence {
			for i := len(changes) - 1; i >= 0; i-- {
				if changes[i].Sequence == firstSequence {
					if excess := i + len(log) - ChannelCacheMaxLength; excess > 0 {
						changes = changes[excess:]
						changesValidFrom = changes[0].Sequence
						i -= excess
					}
					if i > 0 {
						newLog := make(LogEntries, 0, i+len(log))
						newLog = append(newLog, changes[0:i]...)
						newLog = append(newLog, log...)
						c.logs = newLog
						base.LogTo("Cache", "  Added %d entries from view (#%d--#%d) to cache of %q",
							i, changes[0].Sequence, changes[i-1].Sequence, c.channelName)
					}
					c.validFrom = changesValidFrom
					return i
				}
			}
		}
		return 0
	}
}

// Saves a channel log, _if_ there isn't already one in the database.
func (c *channelLogWriter) addChangeLog_(log *channels.ChangeLog) (added bool, err error) {
	added, err = c.bucket.AddRaw(channelLogDocID(c.channelName), 0, encodeChannelLog(log))
	if added {
		base.LogTo("ChannelLog", "Added missing channel-log %q with %d entries",
			c.channelName, log.Len())
	} else {
		base.LogTo("ChannelLog", "Didn't add channel-log %q with %d entries (err=%v)",
			c.channelName, log.Len())
	}
	return
}

func (db *Database) getOldRevisionJSON(docid string, revid string) ([]byte, error) {
	data, err := db.Bucket.GetRaw(oldRevisionKey(docid, revid))
	if base.IsDocNotFoundError(err) {
		base.LogTo("CRUD+", "No old revision %q / %q", docid, revid)
		err = nil
	}
	if data != nil {
		base.LogTo("CRUD+", "Got old revision %q / %q --> %d bytes", docid, revid, len(data))
	}
	return data, err
}

// Updates the Channels property of a document object with current & past channels.
// Returns the set of channels that have changed (document joined or left in this revision)
func (doc *document) updateChannels(newChannels base.Set) (changedChannels base.Set) {
	var changed []string
	oldChannels := doc.Channels
	if oldChannels == nil {
		oldChannels = ChannelMap{}
		doc.Channels = oldChannels
	} else {
		// Mark every no-longer-current channel as unsubscribed:
		curSequence := doc.Sequence
		for channel, removal := range oldChannels {
			if removal == nil && !newChannels.Contains(channel) {
				oldChannels[channel] = &ChannelRemoval{
					Seq:     curSequence,
					RevID:   doc.CurrentRev,
					Deleted: doc.Deleted}
				changed = append(changed, channel)
			}
		}
	}

	// Mark every current channel as subscribed:
	for channel, _ := range newChannels {
		if value, exists := oldChannels[channel]; value != nil || !exists {
			oldChannels[channel] = nil
			changed = append(changed, channel)
		}
	}
	if changed != nil {
		base.LogTo("CRUD", "\tDoc %q in channels %q", doc.ID, newChannels)
		changedChannels = channels.SetOf(changed...)
	}
	return
}

// Top-level handler call. It's passed a pointer to the specific method to run.
func (h *handler) invoke(method handlerMethod) error {
	base.LogTo("HTTP", "%s %s", h.rq.Method, h.rq.URL)
	h.setHeader("Server", VersionString)

	// If there is a "db" path variable, look up the database context:
	if dbname, ok := h.PathVars()["db"]; ok {
		h.context = h.server.databases[dbname]
		if h.context == nil {
			return &base.HTTPError{http.StatusNotFound, "no such database"}
		}
	}

	// Authenticate; admin handlers can ignore missing credentials
	if err := h.checkAuth(); err != nil {
		if !h.admin {
			return err
		}
	}

	// Now look up the database:
	if h.context != nil {
		var err error
		h.db, err = db.GetDatabase(h.context.dbcontext, h.user)
		if err != nil {
			return err
		}
	}

	return method(h) // Call the actual handler code
}

// Writes the response status code, and if it's an error writes a JSON description to the body.
func (h *handler) writeStatus(status int, message string) {
	if status < 300 {
		h.response.WriteHeader(status)
		h.logStatus(status, message)
		return
	}
	// Got an error:
	var errorStr string
	switch status {
	case http.StatusNotFound:
		errorStr = "not_found"
	case http.StatusConflict:
		errorStr = "conflict"
	default:
		errorStr = http.StatusText(status)
		if errorStr == "" {
			errorStr = fmt.Sprintf("%d", status)
		}
	}

	h.setHeader("Content-Type", "application/json")
	h.response.WriteHeader(status)
	base.LogTo("HTTP", "    --> %d %s", status, message)
	jsonOut, _ := json.Marshal(db.Body{"error": errorStr, "reason": message})
	h.response.Write(jsonOut)
}

func (h *handler) logStatus(status int, message string) {
	if base.LogKeys["HTTP+"] {
		duration := float64(time.Since(h.startTime)) / float64(time.Millisecond)
		base.LogTo("HTTP+", "#%03d:     --> %d %s  (%.1f ms)",
			h.serialNumber, status, message, duration)
	}
}

// Updates the Access property of a document object
func (db *Database) updateDocAccess(doc *document, newAccess channels.AccessMap) (changed bool) {
	for name, access := range doc.Access {
		if access.UpdateAtSequence(newAccess[name], doc.Sequence) {
			if len(access) == 0 {
				delete(doc.Access, name)
			}
			changed = true
			db.invalUserChannels(name)
		}
	}
	for name, access := range newAccess {
		if _, existed := doc.Access[name]; !existed {
			changed = true
			if doc.Access == nil {
				doc.Access = UserAccessMap{}
			}
			doc.Access[name] = access.AtSequence(doc.Sequence)
			db.invalUserChannels(name)
		}
	}
	if changed {
		base.LogTo("Access", "Doc %q grants access: %v", doc.ID, doc.Access)
	}
	return
}

func NewDatabaseContext(dbName string, bucket base.Bucket) (*DatabaseContext, error) {
	context := &DatabaseContext{
		Name:        dbName,
		Bucket:      bucket,
		tapNotifier: sync.NewCond(&sync.Mutex{}),
	}
	var err error
	context.sequences, err = newSequenceAllocator(bucket)
	if err != nil {
		return nil, err
	}
	tapFeed, err := bucket.StartTapFeed(walrus.TapArguments{Backfill: walrus.TapNoBackfill})
	if err != nil {
		return nil, err
	}

	// Start a goroutine to broadcast to the tapNotifier whenever a document changes:
	go func() {
		for event := range tapFeed.Events() {
			if event.Opcode == walrus.TapMutation || event.Opcode == walrus.TapDeletion {
				key := string(event.Key)
				if strings.HasPrefix(key, "_sync:") && !strings.HasPrefix(key, "_sync:user") &&
					!strings.HasPrefix(key, "_sync:role") {
					continue // ignore metadata docs (sequence counter, attachments, local docs...)
				}
				base.LogTo("Changes", "Notifying that %q changed (key=%q)", dbName, event.Key)
				context.tapNotifier.Broadcast()
			}
		}
	}()

	return context, nil
}

func (c *changeCache) processPrincipalDoc(docID string, docJSON []byte, isUser bool) {
	// Currently the cache isn't really doing much with user docs; mostly it needs to know about
	// them because they have sequence numbers, so without them the sequence of sequences would
	// have gaps in it, causing later sequences to get stuck in the queue.
	princ, err := c.context.Authenticator().UnmarshalPrincipal(docJSON, "", 0, isUser)
	if princ == nil {
		base.Warn("changeCache: Error unmarshaling doc %q: %v", docID, err)
		return
	}
	sequence := princ.Sequence()
	if sequence <= c.initialSequence {
		return // Tap is sending us an old value from before I started up; ignore it
	}

	// Now add the (somewhat fictitious) entry:
	change := &LogEntry{
		Sequence:     sequence,
		TimeReceived: time.Now(),
	}
	if isUser {
		change.DocID = "_user/" + princ.Name()
	} else {
		change.DocID = "_role/" + princ.Name()
	}

	base.LogTo("Cache", "Received #%d (%q)", change.Sequence, change.DocID)

	c.processEntry(change)
}

// HTTP handler for _dump
func (h *handler) handleDump() error {
	viewName := h.PathVars()["view"]
	base.LogTo("HTTP", "Dump view %q", viewName)
	opts := db.Body{"stale": false, "reduce": false}
	result, err := h.db.Bucket.View("sync_gateway", viewName, opts)
	if err != nil {
		return err
	}
	title := fmt.Sprintf("/%s: “%s” View", html.EscapeString(h.db.Name), html.EscapeString(viewName))
	h.setHeader("Content-Type", `text/html; charset="UTF-8"`)
	h.response.Write([]byte(fmt.Sprintf(
		`<!DOCTYPE html><html><head><title>%s</title></head><body>
		<h1>%s</h1><code>
		<table border=1>
		`,
		title, title)))
	h.response.Write([]byte("\t<tr><th>Key</th><th>Value</th><th>ID</th></tr>\n"))
	for _, row := range result.Rows {
		key, _ := json.Marshal(row.Key)
		value, _ := json.Marshal(row.Value)
		h.response.Write([]byte(fmt.Sprintf("\t<tr><td>%s</td><td>%s</td><td><em>%s</em></td>",
			html.EscapeString(string(key)), html.EscapeString(string(value)), html.EscapeString(row.ID))))
		h.response.Write([]byte("</tr>\n"))
	}
	h.response.Write([]byte("</table>\n</code></html></body>"))
	return nil
}

func (db *Database) AddToChangeLogs(changedChannels base.Set, channelMap ChannelMap, entry channels.LogEntry, parentRevID string) error {
	var err error
	base.LogTo("Changes", "Updating #%d %q/%q in channels %s", entry.Sequence, entry.DocID, entry.RevID, changedChannels)
	for channel, removal := range channelMap {
		if removal != nil && removal.Seq != entry.Sequence {
			continue
		}
		// Set Removed flag appropriately for this channel:
		if removal != nil {
			entry.Flags |= channels.Removed
		} else {
			entry.Flags = entry.Flags &^ channels.Removed
		}
		if err1 := db.AddToChangeLog(channel, entry, parentRevID); err != nil {
			err = err1
		}
	}

	// Finally, add to the universal "*" channel.
	if EnableStarChannelLog {
		entry.Flags = entry.Flags &^ channels.Removed
		if err1 := db.AddToChangeLog("*", entry, parentRevID); err != nil {
			err = err1
		}
	}

	return err
}

// Updates a document's channel/role UserAccessMap with new access settings from an AccessMap.
// Returns an array of the user/role names whose access has changed as a result.
func (accessMap *UserAccessMap) updateAccess(doc *document, newAccess channels.AccessMap) (changedUsers []string) {
	// Update users already appearing in doc.Access:
	for name, access := range *accessMap {
		if access.UpdateAtSequence(newAccess[name], doc.Sequence) {
			if len(access) == 0 {
				delete(*accessMap, name)
			}
			changedUsers = append(changedUsers, name)
		}
	}
	// Add new users who are in newAccess but not accessMap:
	for name, access := range newAccess {
		if _, existed := (*accessMap)[name]; !existed {
			if *accessMap == nil {
				*accessMap = UserAccessMap{}
			}
			(*accessMap)[name] = channels.AtSequence(access, doc.Sequence)
			changedUsers = append(changedUsers, name)
		}
	}
	if changedUsers != nil {
		what := "channel"
		if accessMap == &doc.RoleAccess {
			what = "role"
		}
		base.LogTo("Access", "Doc %q grants %s access: %v", doc.ID, what, *accessMap)
	}
	return changedUsers
}

// HTTP handler for _dumpchannel
func (h *handler) handleDumpChannel() error {
	channelName := h.PathVar("channel")
	since := h.getIntQuery("since", 0)
	base.LogTo("HTTP", "Dump channel %q", channelName)

	chanLog, err := h.db.GetChangeLog(channelName, since)
	if err != nil {
		return err
	} else if chanLog == nil {
		return base.HTTPErrorf(http.StatusNotFound, "no such channel")
	}
	title := fmt.Sprintf("/%s: “%s” Channel", html.EscapeString(h.db.Name), html.EscapeString(channelName))
	h.setHeader("Content-Type", `text/html; charset="UTF-8"`)
	h.response.Write([]byte(fmt.Sprintf(
		`<!DOCTYPE html><html><head><title>%s</title></head><body>
		<h1>%s</h1><code>
		<p>Since = %d</p>
		<table border=1>
		`,
		title, title, chanLog.Since)))
	h.response.Write([]byte("\t<tr><th>Seq</th><th>Doc</th><th>Rev</th><th>Flags</th></tr>\n"))
	for _, entry := range chanLog.Entries {
		h.response.Write([]byte(fmt.Sprintf("\t<tr><td>%d</td><td>%s</td><td>%s</td><td>%08b</td>",
			entry.Sequence,
			html.EscapeString(entry.DocID), html.EscapeString(entry.RevID), entry.Flags)))
		h.response.Write([]byte("</tr>\n"))
	}
	h.response.Write([]byte("</table>\n</code></html></body>"))
	return nil
}