siyuan/kernel/model/repository.go
2022-07-09 09:45:58 +08:00

675 lines
17 KiB
Go

// SiYuan - Build Your Eternal Digital Garden
// Copyright (c) 2020-present, b3log.org
//
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program. If not, see <https://www.gnu.org/licenses/>.
package model
import (
"crypto/rand"
"encoding/base64"
"errors"
"fmt"
"math"
"os"
"strings"
"time"
"github.com/88250/gulu"
"github.com/siyuan-note/dejavu"
"github.com/siyuan-note/encryption"
"github.com/siyuan-note/eventbus"
"github.com/siyuan-note/filelock"
"github.com/siyuan-note/siyuan/kernel/cache"
"github.com/siyuan-note/siyuan/kernel/sql"
"github.com/siyuan-note/siyuan/kernel/util"
)
func init() {
subscribeEvents()
}
func GetRepoSnapshots(page int) (logs []*dejavu.Log, pageCount, totalCount int, err error) {
if 1 > len(Conf.Repo.Key) {
err = errors.New(Conf.Language(26))
return
}
repo, err := newRepository()
if nil != err {
return
}
logs, pageCount, totalCount, err = repo.GetIndexLogs(page, 32)
if nil != err {
if dejavu.ErrNotFoundIndex == err {
logs = []*dejavu.Log{}
err = nil
return
}
util.LogErrorf("get data repo index logs failed: %s", err)
return
}
return
}
func ImportRepoKey(base64Key string) (err error) {
util.PushMsg(Conf.Language(136), 3000)
base64Key = strings.TrimSpace(base64Key)
base64Key = gulu.Str.RemoveInvisible(base64Key)
if 1 > len(base64Key) {
err = errors.New(Conf.Language(142))
return
}
key, err := base64.StdEncoding.DecodeString(base64Key)
if nil != err {
util.LogErrorf("import data repo key failed: %s", err)
return errors.New(Conf.Language(157))
}
Conf.Repo.Key = key
Conf.Save()
if err = os.RemoveAll(Conf.Repo.GetSaveDir()); nil != err {
return
}
if err = os.MkdirAll(Conf.Repo.GetSaveDir(), 0755); nil != err {
return
}
time.Sleep(1 * time.Second)
util.PushMsg(Conf.Language(138), 3000)
time.Sleep(1 * time.Second)
if initErr := IndexRepo("[Init] Init data repo"); nil != initErr {
util.PushErrMsg(fmt.Sprintf(Conf.Language(140), initErr), 0)
}
return
}
func ResetRepo() (err error) {
msgId := util.PushMsg(Conf.Language(144), 1000*60)
if err = os.RemoveAll(Conf.Repo.GetSaveDir()); nil != err {
return
}
if err = os.MkdirAll(Conf.Repo.GetSaveDir(), 0755); nil != err {
return
}
Conf.Repo.Key = nil
Conf.Save()
util.PushUpdateMsg(msgId, Conf.Language(145), 3000)
return
}
func InitRepoKey() (err error) {
util.PushMsg(Conf.Language(136), 3000)
if err = os.RemoveAll(Conf.Repo.GetSaveDir()); nil != err {
return
}
if err = os.MkdirAll(Conf.Repo.GetSaveDir(), 0755); nil != err {
return
}
randomBytes := make([]byte, 16)
_, err = rand.Read(randomBytes)
if nil != err {
return
}
password := string(randomBytes)
randomBytes = make([]byte, 16)
_, err = rand.Read(randomBytes)
if nil != err {
util.LogErrorf("init data repo key failed: %s", err)
return
}
salt := string(randomBytes)
key, err := encryption.KDF(password, salt)
if nil != err {
util.LogErrorf("init data repo key failed: %s", err)
return
}
Conf.Repo.Key = key
Conf.Save()
time.Sleep(1 * time.Second)
util.PushMsg(Conf.Language(138), 3000)
time.Sleep(1 * time.Second)
if initErr := IndexRepo("[Init] Init data repo"); nil != initErr {
util.PushErrMsg(fmt.Sprintf(Conf.Language(140), initErr), 0)
}
return
}
func CheckoutRepo(id string) (err error) {
if 1 > len(Conf.Repo.Key) {
err = errors.New(Conf.Language(26))
return
}
repo, err := newRepository()
if nil != err {
return
}
util.PushEndlessProgress(Conf.Language(63))
writingDataLock.Lock()
defer writingDataLock.Unlock()
WaitForWritingFiles()
sql.WaitForWritingDatabase()
filelock.ReleaseAllFileLocks()
CloseWatchAssets()
defer WatchAssets()
// 恢复快照时自动暂停同步,避免刚刚恢复后的数据又被同步覆盖
syncEnabled := Conf.Sync.Enabled
Conf.Sync.Enabled = false
Conf.Save()
_, _, err = repo.Checkout(id, map[string]interface{}{dejavu.CtxPushMsg: dejavu.CtxPushMsgToStatusBarAndProgress})
if nil != err {
util.PushClearProgress()
return
}
RefreshFileTree()
if syncEnabled {
func() {
time.Sleep(5 * time.Second)
util.PushMsg(Conf.Language(134), 0)
}()
}
return
}
func DownloadCloudSnapshot(tag, id string) (err error) {
if 1 > len(Conf.Repo.Key) {
err = errors.New(Conf.Language(26))
return
}
repo, err := newRepository()
if nil != err {
return
}
cloudInfo, err := buildCloudInfo()
if nil != err {
return
}
defer util.PushClearProgress()
downloadFileCount, downloadChunkCount, downloadBytes, err := repo.DownloadTagIndex(tag, id, cloudInfo, map[string]interface{}{dejavu.CtxPushMsg: dejavu.CtxPushMsgToStatusBarAndProgress})
if nil != err {
return
}
msg := fmt.Sprintf(Conf.Language(153), downloadFileCount, downloadChunkCount, byteCountSI(downloadBytes))
util.PushMsg(msg, 5000)
util.PushStatusBar(msg)
return
}
func UploadCloudSnapshot(tag, id string) (err error) {
if 1 > len(Conf.Repo.Key) {
err = errors.New(Conf.Language(26))
return
}
repo, err := newRepository()
if nil != err {
return
}
cloudInfo, err := buildCloudInfo()
if nil != err {
return
}
util.PushEndlessProgress(Conf.Language(116))
defer util.PushClearProgress()
uploadFileCount, uploadChunkCount, uploadBytes, err := repo.UploadTagIndex(tag, id, cloudInfo, map[string]interface{}{dejavu.CtxPushMsg: dejavu.CtxPushMsgToStatusBarAndProgress})
if nil != err {
if errors.Is(err, dejavu.ErrCloudBackupCountExceeded) {
err = errors.New(Conf.Language(154))
}
return
}
msg := fmt.Sprintf(Conf.Language(152), uploadFileCount, uploadChunkCount, byteCountSI(uploadBytes))
util.PushMsg(msg, 5000)
util.PushStatusBar(msg)
return
}
func RemoveCloudRepoTag(tag string) (err error) {
if 1 > len(Conf.Repo.Key) {
err = errors.New(Conf.Language(26))
return
}
if "" == tag {
err = errors.New("tag is empty")
return
}
repo, err := newRepository()
if nil != err {
return
}
cloudInfo, err := buildCloudInfo()
if nil != err {
return
}
err = repo.RemoveCloudRepoTag(tag, cloudInfo, map[string]interface{}{dejavu.CtxPushMsg: dejavu.CtxPushMsgToStatusBar})
if nil != err {
return
}
return
}
func GetCloudRepoTagSnapshots() (ret []*dejavu.Log, err error) {
if 1 > len(Conf.Repo.Key) {
err = errors.New(Conf.Language(26))
return
}
repo, err := newRepository()
if nil != err {
return
}
cloudInfo, err := buildCloudInfo()
if nil != err {
return
}
ret, err = repo.GetCloudRepoTagLogs(cloudInfo, map[string]interface{}{dejavu.CtxPushMsg: dejavu.CtxPushMsgToStatusBar})
if 1 > len(ret) {
ret = []*dejavu.Log{}
}
return
}
func GetTagSnapshots() (ret []*dejavu.Log, err error) {
if 1 > len(Conf.Repo.Key) {
err = errors.New(Conf.Language(26))
return
}
repo, err := newRepository()
if nil != err {
return
}
ret, err = repo.GetTagLogs()
if 1 > len(ret) {
ret = []*dejavu.Log{}
}
return
}
func RemoveTagSnapshot(tag string) (err error) {
if 1 > len(Conf.Repo.Key) {
err = errors.New(Conf.Language(26))
return
}
repo, err := newRepository()
if nil != err {
return
}
err = repo.RemoveTag(tag)
return
}
func TagSnapshot(id, name string) (err error) {
if 1 > len(Conf.Repo.Key) {
err = errors.New(Conf.Language(26))
return
}
name = gulu.Str.RemoveInvisible(name)
if "" == name {
err = errors.New(Conf.Language(142))
return
}
if !gulu.File.IsValidFilename(name) {
err = errors.New(Conf.Language(151))
return
}
repo, err := newRepository()
if nil != err {
return
}
index, err := repo.GetIndex(id)
if nil != err {
return
}
if err = repo.AddTag(index.ID, name); nil != err {
msg := fmt.Sprintf("Add tag to data snapshot [%s] failed: %s", index.ID, err)
util.PushStatusBar(msg)
return
}
return
}
func IndexRepo(memo string) (err error) {
if 1 > len(Conf.Repo.Key) {
err = errors.New(Conf.Language(26))
return
}
memo = gulu.Str.RemoveInvisible(memo)
if "" == memo {
err = errors.New(Conf.Language(142))
return
}
repo, err := newRepository()
if nil != err {
return
}
util.PushEndlessProgress(Conf.Language(143))
writingDataLock.Lock()
defer writingDataLock.Unlock()
start := time.Now()
latest, _ := repo.Latest()
WaitForWritingFiles()
filelock.ReleaseAllFileLocks()
index, err := repo.Index(memo, map[string]interface{}{
dejavu.CtxPushMsg: dejavu.CtxPushMsgToStatusBarAndProgress,
})
if nil != err {
util.PushStatusBar("Index data repo failed: " + err.Error())
return
}
elapsed := time.Since(start)
if nil == latest || latest.ID != index.ID {
msg := fmt.Sprintf(Conf.Language(147), elapsed.Seconds())
util.PushStatusBar(msg)
util.PushMsg(msg, 5000)
} else {
msg := fmt.Sprintf(Conf.Language(148), elapsed.Seconds())
util.PushStatusBar(msg)
util.PushMsg(msg, 5000)
}
util.PushClearProgress()
return
}
func syncRepo(boot, exit, byHand bool) {
if 1 > len(Conf.Repo.Key) {
msg := Conf.Language(26)
util.PushStatusBar(msg)
util.PushErrMsg(msg, 0)
return
}
repo, err := newRepository()
if nil != err {
msg := fmt.Sprintf("sync repo failed: %s", err)
util.LogErrorf(msg)
util.PushStatusBar(msg)
util.PushErrMsg(msg, 0)
return
}
err = indexRepoBeforeCloudSync(repo)
if nil != err {
return
}
start := time.Now()
cloudInfo, err := buildCloudInfo()
if nil != err {
return
}
syncContext := map[string]interface{}{dejavu.CtxPushMsg: dejavu.CtxPushMsgToStatusBar}
_, mergeResult, trafficStat, err := repo.Sync(cloudInfo, syncContext)
elapsed := time.Since(start)
if nil != err {
util.LogErrorf("sync data repo failed: %s", err)
msg := fmt.Sprintf(Conf.Language(80), formatErrorMsg(err))
if errors.Is(err, dejavu.ErrCloudStorageSizeExceeded) {
msg = fmt.Sprintf(Conf.Language(43), byteCountSI(int64(Conf.User.UserSiYuanRepoSize)))
}
Conf.Sync.Stat = msg
util.PushStatusBar(msg)
util.PushErrMsg(msg, 0)
if boot {
BootSyncSucc = 1
}
if exit {
ExitSyncSucc = 1
}
planSyncAfter(fixSyncInterval)
return
}
util.PushStatusBar(fmt.Sprintf(Conf.Language(149), elapsed.Seconds()))
Conf.Sync.Synced = util.CurrentTimeMillis()
msg := fmt.Sprintf(Conf.Language(150), trafficStat.UploadFileCount, trafficStat.DownloadFileCount, trafficStat.UploadChunkCount, trafficStat.DownloadChunkCount, byteCountSI(trafficStat.UploadBytes), byteCountSI(trafficStat.DownloadBytes))
Conf.Sync.Stat = msg
if 1 > len(mergeResult.Upserts) && 1 > len(mergeResult.Removes) { // 没有数据变更
syncSameCount++
if 10 < syncSameCount {
syncSameCount = 5
}
if !byHand {
delay := time.Minute * time.Duration(int(math.Pow(2, float64(syncSameCount))))
if fixSyncInterval.Minutes() > delay.Minutes() {
delay = time.Minute * 8
}
planSyncAfter(delay)
}
util.PushClearProgress()
return
}
// 有数据变更,需要重建索引
var upserts, removes []string
for _, file := range mergeResult.Upserts {
upserts = append(upserts, file.Path)
}
for _, file := range mergeResult.Removes {
removes = append(removes, file.Path)
}
incReindex(upserts, removes)
cache.ClearDocsIAL()
// 刷新界面
util.ReloadUI()
elapsed = time.Since(start)
go func() {
time.Sleep(2 * time.Second)
util.PushStatusBar(fmt.Sprintf(Conf.Language(149), elapsed.Seconds()))
}()
return
}
func indexRepoBeforeCloudSync(repo *dejavu.Repo) (err error) {
start := time.Now()
latest, _ := repo.Latest()
index, err := repo.Index("[Sync] Cloud sync", map[string]interface{}{
dejavu.CtxPushMsg: dejavu.CtxPushMsgToStatusBar,
})
if nil != err {
msg := fmt.Sprintf(Conf.Language(140), err)
util.PushStatusBar(msg)
util.PushErrMsg(msg, 5000)
util.LogErrorf("index data repo before cloud sync failed: %s", err)
return
}
elapsed := time.Since(start)
if nil == latest || latest.ID != index.ID {
// 对新创建的快照需要更新备注,加入耗时统计
index.Memo = fmt.Sprintf("[Sync] Cloud sync, completed in %.2fs", elapsed.Seconds())
if err = repo.PutIndex(index); nil != err {
util.PushStatusBar("Save data snapshot for cloud sync failed")
util.LogErrorf("put index into data repo before cloud sync failed: %s", err)
return
}
util.PushStatusBar(fmt.Sprintf(Conf.Language(147), elapsed.Seconds()))
} else {
util.PushStatusBar(fmt.Sprintf(Conf.Language(148), elapsed.Seconds()))
}
if 7000 < elapsed.Milliseconds() {
util.LogWarnf("index data repo before cloud sync elapsed [%dms]", elapsed.Milliseconds())
}
return
}
func newRepository() (ret *dejavu.Repo, err error) {
ignoreLines := getIgnoreLines()
ignoreLines = append(ignoreLines, "/.siyuan/conf.json") // 忽略旧版同步配置
ret, err = dejavu.NewRepo(util.DataDir, util.RepoDir, util.HistoryDir, util.TempDir, Conf.Repo.Key, ignoreLines)
if nil != err {
util.LogErrorf("init data repo failed: %s", err)
}
return
}
func subscribeEvents() {
eventbus.Subscribe(dejavu.EvtIndexWalkData, func(context map[string]interface{}, path string) {
msg := "Indexing data repo [walk data " + path + "]"
util.SetBootDetails(msg)
contextPushMsg(context, msg)
})
eventbus.Subscribe(dejavu.EvtIndexGetLatestFile, func(context map[string]interface{}, path string) {
msg := "Indexing data repo [get latest file " + path + "]"
util.SetBootDetails(msg)
contextPushMsg(context, msg)
})
eventbus.Subscribe(dejavu.EvtIndexUpsertFile, func(context map[string]interface{}, path string) {
msg := "Indexing data repo [upsert file " + path + "]"
util.SetBootDetails(msg)
contextPushMsg(context, msg)
})
eventbus.Subscribe(dejavu.EvtCheckoutWalkData, func(context map[string]interface{}, path string) {
msg := "Checkout data repo [walk data " + path + "]"
util.SetBootDetails(msg)
contextPushMsg(context, msg)
})
eventbus.Subscribe(dejavu.EvtCheckoutUpsertFile, func(context map[string]interface{}, path string) {
msg := "Checkout data repo [upsert file " + path + "]"
util.SetBootDetails(msg)
contextPushMsg(context, msg)
})
eventbus.Subscribe(dejavu.EvtCheckoutRemoveFile, func(context map[string]interface{}, path string) {
msg := "Checkout data repo [remove file " + path + "]"
util.SetBootDetails(msg)
contextPushMsg(context, msg)
})
eventbus.Subscribe(dejavu.EvtCloudBeforeDownloadIndex, func(context map[string]interface{}, id string) {
msg := "Downloading data repo index [" + id + "]"
util.SetBootDetails(msg)
contextPushMsg(context, msg)
})
eventbus.Subscribe(dejavu.EvtCloudBeforeDownloadFile, func(context map[string]interface{}, id string) {
msg := "Downloading data repo file [" + id + "]"
util.SetBootDetails(msg)
contextPushMsg(context, msg)
})
eventbus.Subscribe(dejavu.EvtCloudBeforeDownloadChunk, func(context map[string]interface{}, id string) {
msg := "Downloading data repo chunk [" + id + "]"
util.SetBootDetails(msg)
contextPushMsg(context, msg)
})
eventbus.Subscribe(dejavu.EvtCloudBeforeDownloadRef, func(context map[string]interface{}, ref string) {
msg := "Downloading data repo ref [" + ref + "]"
util.SetBootDetails(msg)
contextPushMsg(context, msg)
})
eventbus.Subscribe(dejavu.EvtCloudBeforeUploadIndex, func(context map[string]interface{}, id string) {
msg := "Uploading data repo index [" + id + "]"
util.SetBootDetails(msg)
contextPushMsg(context, msg)
})
eventbus.Subscribe(dejavu.EvtCloudBeforeUploadFile, func(context map[string]interface{}, id string) {
msg := "Uploading data repo file [" + id + "]"
util.SetBootDetails(msg)
contextPushMsg(context, msg)
})
eventbus.Subscribe(dejavu.EvtCloudBeforeUploadChunk, func(context map[string]interface{}, id string) {
msg := "Uploading data repo chunk [" + id + "]"
util.SetBootDetails(msg)
contextPushMsg(context, msg)
})
eventbus.Subscribe(dejavu.EvtCloudBeforeUploadRef, func(context map[string]interface{}, ref string) {
msg := "Uploading data repo ref [" + ref + "]"
util.SetBootDetails(msg)
contextPushMsg(context, msg)
})
}
func contextPushMsg(context map[string]interface{}, msg string) {
switch context[dejavu.CtxPushMsg].(int) {
case dejavu.CtxPushMsgToProgress:
util.PushEndlessProgress(msg)
case dejavu.CtxPushMsgToStatusBar:
util.PushStatusBar(msg)
case dejavu.CtxPushMsgToStatusBarAndProgress:
util.PushStatusBar(msg)
util.PushEndlessProgress(msg)
}
}
func buildCloudInfo() (ret *dejavu.CloudInfo, err error) {
if !IsValidCloudDirName(Conf.Sync.CloudName) {
util.LogWarnf("invalid cloud repo name, rename it to [main]")
Conf.Sync.CloudName = "main"
Conf.Save()
}
if nil == Conf.User {
err = errors.New("user auth failed")
return
}
ret = &dejavu.CloudInfo{
Dir: Conf.Sync.CloudName,
UserID: Conf.User.UserId,
Token: Conf.User.UserToken,
LimitSize: int64(Conf.User.UserSiYuanRepoSize - Conf.User.UserSiYuanAssetSize),
ProxyURL: Conf.System.NetworkProxy.String(),
Server: util.AliyunServer,
}
return
}