func pushChanges(
	clientInfo *db.ClientInfo,
	docInfo *db.DocInfo,
	pack *change.Pack,
	initialServerSeq uint64,
) (*checkpoint.Checkpoint, []*change.Change, error) {
	cp := clientInfo.Checkpoint(docInfo.ID)
	var pushedChanges []*change.Change
	for _, c := range pack.Changes {
		if c.ID().ClientSeq() > cp.ClientSeq {
			serverSeq := docInfo.IncreaseServerSeq()
			cp = cp.NextServerSeq(serverSeq)
			c.SetServerSeq(serverSeq)
			pushedChanges = append(pushedChanges, c)
		} else {
			log.Logger.Warnf("change is rejected: %d vs %d ", c.ID().ClientSeq(), cp.ClientSeq)
		}
		cp = cp.SyncClientSeq(c.ClientSeq())
	}
	if len(pack.Changes) > 0 {
		log.Logger.Infof(
			"PUSH: '%s' pushes %d changes into '%s', rejected %d changes, serverSeq: %d -> %d, cp: %s",
			clientInfo.ID,
			len(pushedChanges),
			docInfo.Key,
			len(pack.Changes)-len(pushedChanges),
			initialServerSeq,
			docInfo.ServerSeq,
			cp.String(),
		)
	}
	return cp, pushedChanges, nil
}
pushChanges returns the changes excluding already saved in DB.
package packs
import (
	"context"
	"errors"
	"fmt"
	gotime "time"
	"github.com/yorkie-team/yorkie/api/converter"
	"github.com/yorkie-team/yorkie/internal/log"
	"github.com/yorkie-team/yorkie/pkg/document"
	"github.com/yorkie-team/yorkie/pkg/document/change"
	"github.com/yorkie-team/yorkie/pkg/document/checkpoint"
	"github.com/yorkie-team/yorkie/pkg/document/key"
	"github.com/yorkie-team/yorkie/pkg/document/time"
	"github.com/yorkie-team/yorkie/pkg/types"
	"github.com/yorkie-team/yorkie/yorkie/backend"
	"github.com/yorkie-team/yorkie/yorkie/backend/db"
	"github.com/yorkie-team/yorkie/yorkie/backend/sync"
)
var (
	ErrInvalidServerSeq = errors.New("invalid server seq")
)
func NewPushPullKey(documentKey *key.Key) sync.Key {
	return sync.NewKey(fmt.Sprintf("pushpull-%s", documentKey.BSONKey()))
}
func NewSnapshotKey(documentKey *key.Key) sync.Key {
	return sync.NewKey(fmt.Sprintf("snapshot-%s", documentKey.BSONKey()))
}
func PushPull(
	ctx context.Context,
	be *backend.Backend,
	clientInfo *db.ClientInfo,
	docInfo *db.DocInfo,
	reqPack *change.Pack,
) (*change.Pack, error) {
	initialServerSeq := docInfo.ServerSeq
	pushedCP, pushedChanges, err := pushChanges(clientInfo, docInfo, reqPack, initialServerSeq)
	if err != nil {
		return nil, err
	}
	respPack, err := pullPack(ctx, be, clientInfo, docInfo, reqPack, pushedCP, initialServerSeq)
	if err != nil {
		return nil, err
	}
	if err := clientInfo.UpdateCheckpoint(docInfo.ID, respPack.Checkpoint); err != nil {
		return nil, err
	}
	if len(pushedChanges) > 0 {
		if err := be.DB.StoreChangeInfos(ctx, docInfo, initialServerSeq, pushedChanges); err != nil {
			return nil, err
		}
	}
	if err := be.DB.UpdateClientInfoAfterPushPull(ctx, clientInfo, docInfo); err != nil {
		return nil, err
	}
	respPack.MinSyncedTicket, err = be.DB.UpdateAndFindMinSyncedTicket(
		ctx,
		clientInfo,
		docInfo.ID,
		reqPack.Checkpoint.ServerSeq,
	)
	if err != nil {
		return nil, err
	}
	if reqPack.HasChanges() {
		be.AttachGoroutine(func() {
			publisherID, err := time.ActorIDFromHex(clientInfo.ID.String())
			if err != nil {
				log.Logger.Error(err)
				return
			}
			ctx := context.Background()
			locker, err := be.Coordinator.NewLocker(
				ctx,
				NewSnapshotKey(reqPack.DocumentKey),
			)
			if err != nil {
				log.Logger.Error(err)
				return
			}
			if err := locker.Lock(ctx); err != nil {
				log.Logger.Error(err)
				return
			}
			defer func() {
				if err := locker.Unlock(ctx); err != nil {
					log.Logger.Error(err)
					return
				}
			}()
			be.Coordinator.Publish(
				ctx,
				publisherID,
				sync.DocEvent{
					Type:         types.DocumentsChangedEvent,
					Publisher:    types.Client{ID: publisherID},
					DocumentKeys: []*key.Key{reqPack.DocumentKey},
				},
			)
			if err := storeSnapshot(
				ctx,
				be,
				docInfo,
			); err != nil {
				log.Logger.Error(err)
			}
		})
	}
	return respPack, nil
}