1 module Rakka.Storage.Impl
9 import Control.Concurrent
10 import Control.Concurrent.STM
11 import Control.Exception
15 import qualified Data.Set as S
16 import Network.HTTP.Lucu
19 import Rakka.Storage.DefaultPage
20 import Rakka.Storage.Repos
21 import Rakka.Storage.Types
22 import Subversion.Types
23 import Subversion.FileSystem
24 import Subversion.Repository
25 import System.Directory
26 import System.FilePath
28 import System.Log.Logger
29 import System.Posix.Files
30 import System.Posix.Types
31 import System.Posix.IO
32 import Text.HyperEstraier hiding (WriteLock)
36 logger = "Rakka.Storage"
39 getPage' :: Repository -> PageName -> Maybe RevNum -> IO (Maybe Page)
40 getPage' repos name rev
41 = do page <- loadPageInRepository repos name rev
43 Nothing -> loadDefaultPage name
47 putPage' :: Repository -> Page -> IO StatusCode
48 putPage' = putPageIntoRepository
51 deletePage' :: Repository -> PageName -> IO StatusCode
52 deletePage' = deletePageFromRepository
55 findAllPages :: Repository -> RevNum -> IO (Set PageName)
56 findAllPages _ 0 = findAllDefaultPages
57 findAllPages repos rev = do reposPages <- findAllPagesInRevision repos rev
58 defaultPages <- findAllDefaultPages
59 return (reposPages `S.union` defaultPages)
62 findChangedPages :: Repository -> RevNum -> RevNum -> IO (Set PageName)
63 findChangedPages repos 0 newRev = findAllPages repos newRev
64 findChangedPages repos oldRev newRev
65 = mapM (findChangedPagesAtRevision repos) [oldRev + 1 .. newRev]
70 getCurrentRevNum :: Repository -> IO RevNum
71 getCurrentRevNum repos
72 = getRepositoryFS repos >>= getYoungestRev
75 startIndexManager :: FilePath -> Repository -> (Page -> IO Document) -> IO (TChan IndexReq)
76 startIndexManager lsdir repos mkDraft
77 = do chan <- newTChanIO
78 index <- openIndex indexDir revFile
79 forkIO (loop chan index)
82 indexDir = lsdir </> "index"
83 revFile = lsdir </> "indexRev"
85 loop :: TChan IndexReq -> Database -> IO ()
87 = do req <- atomically $ readTChan chan
90 -> do noticeM logger "Rebuilding the H.E. index..."
92 removeDirectoryRecursive indexDir
93 index' <- openIndex indexDir revFile
94 syncIndex' index' revFile repos mkDraft
98 -> do syncIndex' index revFile repos mkDraft
102 -> do result <- searchIndex index cond
103 atomically $ putTMVar var result
107 -- casket を R/W モードで開く。成功したらそのまま返し、失敗したら
108 -- indexDir と revFile を削除してから casket を R/W モードで開く。
109 openIndex :: FilePath -> FilePath -> IO Database
110 openIndex indexDir revFile
111 = do ret <- openDatabase indexDir (Writer [])
114 -> do debugM logger ("Opened an H.E. index on " ++ indexDir)
118 -> do noticeM logger ("Failed to open an H.E. index on "
119 ++ indexDir ++ ": " ++ show err)
121 indexExists <- doesDirectoryExist indexDir
123 $ removeDirectoryRecursive indexDir
125 revFileExists <- doesFileExist revFile
129 Right index <- openDatabase indexDir (Writer [Create []])
130 addAttrIndex index "@uri" SeqIndex
131 addAttrIndex index "rakka:revision" SeqIndex
132 noticeM logger ("Created an H.E. index on " ++ indexDir)
137 syncIndex' :: Database -> FilePath -> Repository -> (Page -> IO Document) -> IO ()
138 syncIndex' index revFile repos mkDraft
139 = updateIndexRev revFile $ \ oldRev ->
140 do debugM logger ("The index revision is currently " ++ show oldRev)
142 newRev <- getCurrentRevNum repos
143 debugM logger ("The repository revision is currently " ++ show newRev)
145 when (oldRev == 0 || newRev /= oldRev)
146 $ syncIndex'' oldRev newRev
149 syncIndex'' :: RevNum -> RevNum -> IO ()
150 syncIndex'' oldRev newRev
151 = do pages <- findChangedPages repos oldRev newRev
152 mapM_ (updateIndex index repos mkDraft newRev) (S.toList pages)
155 searchIndex :: Database -> Condition -> IO [(PageName, RevNum)]
156 searchIndex index cond
157 = searchDatabase index cond >>= mapM fromId
159 fromId :: DocumentID -> IO (PageName, RevNum)
161 = do uri <- getDocURI index docId
162 rev <- getDocAttr index docId "rakka:revision"
163 >>= return . read . fromJust
164 return (decodePageName $ uriPath uri, rev)
167 updateIndex :: Database
169 -> (Page -> IO Document)
173 updateIndex index repos mkDraft rev name
174 = do pageM <- getPage' repos name (Just rev)
178 -> do docIdM <- getDocIdByURI index (mkRakkaURI name)
181 Just docId -> do removeDocument index docId [CleaningRemove]
182 infoM logger ("Removed page " ++ name ++ " from the index")
184 -> do draft <- mkDraft page
185 putDocument index draft [CleaningPut]
186 infoM logger ("Indexed page " ++ name ++ " of revision " ++ show (pageRevision page))
189 updateIndexRev :: FilePath -> (RevNum -> IO RevNum) -> IO ()
190 updateIndexRev revFile f = bracket acquireLock releaseLock update
194 = do fd <- openFd revFile ReadWrite (Just stdFileMode) defaultFileFlags
195 waitToSetLock fd (WriteLock, AbsoluteSeek, 0, 0)
198 releaseLock :: Fd -> IO ()
200 = setLock fd (Unlock, AbsoluteSeek, 0, 0)
202 update :: Fd -> IO ()
204 = do fdSeek fd AbsoluteSeek 0
205 size <- return . fromIntegral . fileSize =<< getFdStatus fd
206 (revStr, gotSize) <- fdRead fd size
207 when (size /= gotSize) $ fail ("read " ++ show gotSize ++
208 " bytes but expected " ++ show size ++ " bytes")
210 let rev = case revStr of
216 let revStr' = show rev' ++ "\n"
217 size' = fromIntegral $ length revStr'
219 fdSeek fd AbsoluteSeek 0
221 wroteSize <- fdWrite fd revStr'
222 when (size' /= wroteSize) $ fail ("wrote " ++ show wroteSize ++
223 " bytes but expected " ++ show size' ++ " bytes")